流年 ℃

一、安装erlang

下载最新稳定版本，编译安装，例如：

http://erlang.org/download/otp_src_21.1.tar.gz

wget http://erlang.org/download/otp_src_21.1.tar.gz .
tar -xzvf otp_src_21.1.tar.gz
cd otp_src_21.1
./configure --prefix=/data/service/erlang/otp_21.1
sudo make
sudo make install

报错： configure: error: No curses library functions found configure: error: /home/xiao/download/otp_src_21.1/erts/configure failed for erts

缺少ncurses，安装sudo apt install libncurses5-dev

二、安装rabbitmq

下载最新稳定版本，编译安装，例如：

https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.8/rabbitmq-server-generic-unix-3.7.8.tar.xz

xz -d abbitmq-server-generic-unix-3.7.8.tar.xz
tar -xvf rabbitmq-server-generic-unix-3.7.8.tar
mv rabbitmq_server... /data/service/rabbitmq
cd /data/service/rabbitmq
./sbin/rabbitmq-server
//此时可能会报erl找不到
//这时需要把/data/service/erlang/otp_21.1/bin加入到环境变量PATH中去
//
//./sbin/rabbitmq-server -detached

报erl找不到，将erlang目录加入PATH： vim /etc/sudoers，修改：

……

批量杀死进程

ps -ef|grep test.sh
xiao      3296 28517  0 10:09 pts/3    00:00:06 /bin/bash ./test.sh
xiao      3365 28517  0 10:09 pts/3    00:00:06 /bin/bash ./test.sh
xiao      3401 28517  0 10:09 pts/3    00:00:06 /bin/bash ./test.sh
xiao      3443 28517  0 10:09 pts/3    00:00:06 /bin/bash ./test.sh
xiao      3480 28517  0 10:09 pts/3    00:00:06 /bin/bash ./test.sh
xiao      5299 28517  0 10:10 pts/3    00:00:06 /bin/bash ./test.sh
xiao      5396 28517  0 10:10 pts/3    00:00:06 /bin/bash ./test.sh
xiao      5431 28517  0 10:10 pts/3    00:00:06 /bin/bash ./test.sh
xiao      5470 28517  0 10:10 pts/3    00:00:06 /bin/bash ./test.sh
xiao      5488 28517  0 10:10 pts/3    00:00:06 /bin/bash ./test.sh
xiao      6589 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh
xiao      6754 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh
xiao      6899 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh
xiao      6972 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh
xiao      7015 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh
xiao      7095 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh
xiao      7129 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh
xiao      7172 28517  0 10:33 pts/3    00:00:05 /bin/bash ./test.sh


ps -ef|grep test.sh|awk '{print "kill -9 " $2}'|sh

服务启动、重启、停止

一个服务的启动、重启、停止的shell脚步：

……

最近公司的服务器经常因为磁盘被日志文件写爆，磁盘不可用后导致接口请求报错，为此经常要定期删除服务器日志以腾出磁盘空间。有没有更好的办法解决磁盘容量太小，日志保存不能太多、时间不能太长的麻烦呢？我试想过，将接口日志收收集后转储到公司内部服务器，做离线存储方案，这样有三个好处：

1、海量接口日志不用再占用线上服务器的有限磁盘空间，节省费用；

2、在负载均衡多节点的部署架构下，避免日志文件的分散导致的查阅不方便；

3、归总转储离线后，日志可以更长时间的存储，本地集中查阅高效，为进一步分析提供了可能性。 比如：收集日志后，可以统计每日访问的pu、pv，可以分析每个接口请求频次；但业务操作存在bug时，可以借助日志分析问题原因……

但如何转储呢？在网上查阅了一些资料，有人利用kafka消息队列+lua，集中存储nginx请求日志到kafka中。

最近刚好在研究nsq，借此做一个基于nsq+lua的nginx日志转储方案，如下！

一、资源下载：

依赖：

https://github.com/openresty/lua-cjson/archive/2.1.0.6.tar.gz

https://github.com/openresty/lua-resty-core/archive/v0.1.13.tar.gz

lua-nsq：

https://github.com/rainingmaster/lua-resty-nsq

注意将Makefile文件里修改lua相关参数 ：

#LUA_INCLUDE_DIR ?= $(PREFIX)/include
#LUA_LIB_DIR ?=     $(PREFIX)/lib/lua/$(LUA_VERSION)

LUA_INCLUDE_DIR ?= $(PREFIX)/include/luajit-2.0
LUA_LIB_DIR ?=     $(PREFIX)/lib/lua/5.1

二、nginx站点server里配置如下：

body_filter_by_lua '
    local resp_body = string.sub(ngx.arg[1], 1, 1000)
    ngx.ctx.buffered = (ngx.ctx.buffered or"") .. resp_body
';
log_by_lua_file /data/service/tengine/conf/lua/log.lua;

三、log.lua脚步如下：

……

nsq的设计特点

分布式的队列传输节点，能够避免传输节点的单点故障，确保消息一定会被传输一次， 但是这样可能引起消息被多次传输 bound the memory footprint of a single process (by persisting some messages to disk)简化消费者和生产者的配置， 提供了一个简单明确的升级路线，提升了效率。

nsqde的基本概念

• nsqd, 转发消息的核心, 负责接受消息， 将消息排队， 发送给客户端, 不同的nsqd之间的消息不会共享
• nsqlookupd ，只是提供client查询当前所有的nsqd的信息， 不会对实际的消息转发做任何的处理， nsqlookupd于nsqd之间在底层建立了tcp的长连接连接， 而默认监听的4160端口就是干这个用的，通过周期性的报告nsqd的状态，
• client 消费者， 需要配置nsqlookupd的地址
• nsqadmin 供管理员集中查看所有的集群，有一个webui
• topic， 属于nsqd， 就是一个消息流，目测是对业务的标志， 或者是为了将同一个物理管道进行逻辑细分的方式， 算是为了提高效率和细分资源，topic并不需要进行实现的配置，一个nsqd可能拥有多个topic
• channel， 属于nsqd，是发送给一个topic的逻辑组， 一个topic可能用拥有多个channel， 每个channel都能收到这个topic下面的所有的消息， 一般一个channel都会对应一个消费者， 而一个channel可以对应多个消费者，channale会将消息随机发送给下面的一个client，所以任意一个client收到的消息都不是完全的，
• channel和topic都是单独缓存消息的

分布式方案

• nsqd随意起， nsqlookup使用备份的方式
• 一个client只会同时对一个nsqd建立连接， 所以一旦一个nsqd连接， 那么就不会对其他的topic建立连接
• 只有再一个nsqd坏掉的时候，才会重新选择nsqd
• 那么client选择nsqd的原则是什么？

nsq保证消息能够正常至少传输一次的方式是

1. client表明已经可以接受消息。
2. nsqd将消息发送出去， 同时将这个消息进行本地存储。
3. client如果回复FIN 表示成功接受， 如果回复REQ， 表明需要重发， 如果没有回复， 则认为超时了， 进行重发。

所以， 当nsqd异常关闭的时候， 没有来得及保存到本地的消息可能会丢失, 解决办法是讲同样的消息发送到两个nsqd中。

这里如何保证消息不丢失的？

nsdlookup 如何路由请求

{
    "channels": [ "nsq_to_file", "c" ], 
    "producers": [ 
        { 
            "remote_address": "127.0.0.1:58148", 
            "hostname": "safedev01v.add.corp.qihoo.net", 
            "broadcast_address": "safedev01v.add.corp.qihoo.net", 
            "tcp_port": 4150, 
            "http_port": 4151, 
            "version": "0.3.6" 
        }, 
        { 
            "remote_address": "10.16.59.85:39652", 
            "hostname": "safedev02v.add.corp.qihoo.net", 
            "broadcast_address": "safedev02v.add.corp.qihoo.net", 
            "tcp_port": 4150, 
            "http_port": 4151, 
            "version": "0.3.7" 
         } 
     ] 
}

雪崩如何处理

……

排序算法种类繁多。

根据处理的数据规模与存储特点，可分为内部排序和外部排序：前者处理的数据规模不大，内存足以容纳；后者处理的数据规模较大，必须将数据存放于外部存储器中，每次排序的时候需要访问外存。根据输入的不同形式，分为脱机算法和在线算法：前者待排序的数据是以批处理的形式给出的；而在云计算之类的环境中，待排序的数据是实时生成的，在排序算法开始运行时，数据并未完全就绪，而是随着排序算法本身的进行而逐步给出的。另外，针对不同的体系结构，又分为串行和并行两大类排序算法。根据算法是否采用随机策略，还有确定式和随机式之分。

冒泡排序(O(n^2))

冒泡排序是比较相邻两数的大小来完成排序的。这里定义比较边界，也就是进行大小比较的边界。对于长度为n的数组，第一趟的比较边界为[0,n-1]，也就是说从a[0]开始，相邻元素两两比较大小，如果满足条件就进行交换，否则继续比较，一直到最后一个比较的元素为a[n-1]为止，此时第一趟排序完成。以升序排序为例，每趟排序完成之后，比较边界中的最大值就沉入底部，比较边界就向前移动一个位置。所以，第二趟排序开始时，比较边界是[0,n-2]。对于长度为n的序列，最多需要n趟完成排序，所以冒泡排序就由两层循环构成，最外层循环用于控制排序的趟数，最内层循环用于比较相邻数字的大小并在本趟排序完成时更新比较边界。

//冒泡排序
public static void bubbleSort(int[] arr,int len){
    int temp=0;
    int compareRange=len-1;//冒泡排序中，参与比较的数字的边界。
    //冒泡排序主要是比较相邻两个数字的大小，以升序排列为例，如果前侧数字大于后侧数字，就进行交换，一直到比较边界。
    for (int i = 0; i <len ; i++) {//n个数使用冒泡排序，最多需要n趟完成排序。最外层循环用于控制排序趟数
        for (int j = 1; j <=compareRange ; j++) {
            if(arr[j-1]>arr[j]){
                temp=arr[j-1];
                arr[j-1]=arr[j];
                arr[j]=temp;
            }
        }
        compareRange--;//每进行一趟排序，序列中最大数字就沉到底部，比较边界就向前移动一个位置。
    }
    System.out.println("排序后数组"+Arrays.toString(arr));
}

在排序后期可能数组已经有序了而算法却还在一趟趟的比较数组元素大小，可以引入一个标记，如果在一趟排序中，数组元素没有发生过交换说明数组已经有序，跳出循环即可。优化后的代码如下：

……