Redis主从集群+哨兵搭建实战 Redis主从集群+哨兵搭建实战背景当请求量逐渐变大，单机Redis可能撑不住请求的时候就要考虑将Redis做集群，入门级别的Redis集群就是主从集群，利用读写分离的特性提供高可用，引入新的技术就会带来新的问题，当然引入集群也是有很多坏处的，比如业务逻辑变复杂，需要读写分离和负载均衡（redis代理解决）网络复杂，有可能节点之间由于网络不可达产生分区风险（CAP理论）一主多从，主宕机集群部分不可用（哨兵模式解决）需要更多硬件资源支持（废话）当然带来的这些问题就有相应的解决方案，下面一步一步带你入坑。主从集群搭建环境准备docker run -dit --name r2 --privileged centos /usr/sbin/init docker run -dit --name r3 --privileged centos /usr/sbin/init docker run -dit --name r4 --privileged centos /usr/sbin/init它们的IP分别如下，我们的规划是r2作为主，其余为从r2 172.17.0.2 # 主 r3 172.17.0.3 # 从 r4 172.17.0.4 # 从docker查看容器ip的命令为docker inspect --format '{{ .NetworkSettings.IPAddress }}' r2安装Redis进入容器命令docker exec -it r2 bash每台分别安装Redis，采用源码编译安装的方式，安装版本为5.0.5，可自定义REDIS_VERSION=5.0.5 # 编译环境安装和一些常用工具 yum install -y wget gcc make telnet # 创建文件夹 cd ~ && mkdir soft && cd soft # 下载软件 wget http://download.redis.io/releases/redis-$REDIS_VERSION.tar.gz # 解压 tar xf redis-$REDIS_VERSION.tar.gz && cd redis-$REDIS_VERSION # 看README.md 执行出错看缺什么就先装什么，装完之后clean一下(make distclean)再次make # 生产环境做好执行一下测试 make test # 可以查看一下生成的可执行程序xxx.o ls -lh src make # 安装 make install PREFIX=/opt/redis$REDIS_VERSION # 添加环境变量 cat >> /etc/profile <<EOF export REDIS_HOME=/opt/redis$REDIS_VERSION export PATH=$PATH:\$REDIS_HOME/bin EOF echo 'source /etc/profile' >> ~/.bashrc && source ~/.bashrc cd utils # 可以执行—次或多次，需要手动确定参数 # a)一个物理机中可以有多个redis实例(进程)，通过port区分 # b)可执行程序就一份在目录，但是内存中未来的多个实例需要各自的配置文件，持久化目录等资源 # c) service redis_6379 start/stop/stauts>linux /etc/init.d/**** # d)脚本还会帮你启动! ./install_server.sh ps -fe | grep redis更改配置文件中的ipsed -i 's/bind 127.0.0.1/bind 0.0.0.0/g' /etc/redis/6379.conf配置文件位置在/etc/redis/6379.conf，如果要在内网访问，需要修改配置文件的bind 127.0.0.1中的ip为0.0.0.0，意为允许任何连接，否则在局域网中是连不上的，这也是用docker进行网络隔离的好处，可以更贴合生产环境，发现更多问题。设置非守护进程，让redis在前台阻塞运行，测试时方便观察输出，当然生产环境中不建议关闭# 关闭守护进程 sed -i 's/daemonize yes/daemonize no/g' /etc/redis/6379.conf # 关掉日志 # logfile /var/log/redis_6379.log sed -i 's|logfile /var/log/redis_6379.log|#logfile /var/log/redis_6379.log|g' /etc/redis/6379.conf重启下redis# -15优雅的杀掉 kill -15 `ps aux | grep -v grep | grep redis | awk '{ print $2 }'` # 重启 redis-server /etc/redis/6379.conf客户端连接，新开shell窗口连接一下看看是否能连接成功# 查看帮助 redis-cli -h # 指定主机端口连接 redis-cli -h 127.0.0.1 -p 6379启动主从集群上一步每台都安装了redis并且验证了可以启动，那么现在来设置主从集群启动，172.17.0.2为主，可以不用管它，172.17.0.3和172.17.0.4是从，启动的时候需要用replicaof命令指定是从172.17.0.2的复制，在3和4上分别执行如下命令即可redis-server /etc/redis/6379.conf --replicaof 172.17.0.2 6379看到输出如下，说明从机已经连上主机了，随时准备接受请求... * Connecting to MASTER 172.17.0.2:6379 ... * MASTER <-> REPLICA sync started ... * Non blocking connect for SYNC fired the event. ... * Master replied to PING, replication can continue... ... * Partial resynchronization not possible (no cached master) ... * Full resync from master: 3c1b294dd75b88d62a9f6be85593b51264f68970:0 ... * MASTER <-> REPLICA sync: receiving 175 bytes from master ... * MASTER <-> REPLICA sync: Flushing old data ... * MASTER <-> REPLICA sync: Loading DB in memory ... * MASTER <-> REPLICA sync: Finished with success 测试首先，在主上设置一个键，在从上试试能否获取即可# 主 set k1 aaa # 从 get k1如下图，发现主设置值之后从也有值，主删除之后从也获取不到了，说明我们的主从集群搭建成功！并且是读写分离的，从只能读，主读写都可以，在从上设置值会报错，但是也可以通过配置开启从机的写权限slave-read-only no，这样意义并不是很大，因为从机的任务就是数据备份和负载均衡的set k2 bash (error) READONLY You can't write against a read only replica.当然这样的集群是有问题，假如现在主宕机了，那么整个集群就是只读的，并且一部分流向主的读请求也会失败（除非读请求全部流向从机），主挂了之后发现从机一直连不上主... # Connection with master lost. # 和主的连接丢失 ... * Caching the disconnected master state. # 缓存一下状态 ... * Connecting to MASTER 172.17.0.2:6379 # 一直在尝试连接主 ... * MASTER <-> REPLICA sync started ... # Error condition on socket for SYNC: Connection refused ... * Connecting to MASTER 172.17.0.2:6379 ... * MASTER <-> REPLICA sync started ... # Error condition on socket for SYNC: Connection refused 那么有没有解决该问题的办法呢？答案是有的，那就是哨兵机制！哨兵可以监控整个集群的健康状态，当主宕机之后可以将从提升为主，从而实现故障转移！引入哨兵哨兵一般都是奇数个，因为需要过半数表决，这就是CAP理论的知识了，这里不再赘述；哨兵一定不要和Redis服务器在同一台机器上，因为Redis宕机了那么哨兵也就宕机了，起不到监控作用，所以我们新建三个容器用来作为哨兵的角色。一套哨兵是可以监控多个Redis主从集群的哨兵环境准备docker run -dit --name s1 --privileged centos /usr/sbin/init docker run -dit --name s2 --privileged centos /usr/sbin/init docker run -dit --name s3 --privileged centos /usr/sbin/init各节点ip如下s1 172.17.0.5 s2 172.17.0.6 s3 172.17.0.7安装哨兵也是Redis服务器，只不过只是具有发布/订阅功能的Redis服务器，所以所有哨兵都需要先安装Redis，参考最上面Redis的安装章节配置并启动新增哨兵配置文件# port 此哨兵工作的port # 语法 sentinel monitor <master-name> <ip> <redis-port> <quorum> # <master-name> 取个名字，一套哨兵可能监控很多redis集群 # <ip> 目标集群主的那台机器的IP # <redis-port> 目标集群主的那台机器的端口 # <quorum> 比如3个哨兵，有2个意见达成一致产生群体效应，这个quorum就写2 cat > /etc/redis/sentinel7380.conf<<EOF port 7380 sentinel monitor mymaster 172.17.0.2 6379 2 EOFsentinel monitor mymaster 172.17.0.2 6379 2解释mymaster ：随便取个名字，一套哨兵是可以监控多个Redis集群的172.17.0.2：指定主的IP6379：指定主的端口2：quorum，表示有2个意见达成一致就可以产生群体效应，做表决动作在三台上分别启动哨兵redis-server /etc/redis/sentinel7380.conf --sentinel当第一台哨兵启动后他也能知道目前集群中有哪些机器，因为主知道... # +monitor master mymaster 172.17.0.2 6379 quorum 2 ... * +slave slave 172.17.0.3:6379 172.17.0.3 6379 @ mymaster 172.17.0.2 6379 ... * +slave slave 172.17.0.4:6379 172.17.0.4 6379 @ mymaster 172.17.0.2 6379启动剩余的哨兵，发现最先启动的哨兵知道后面两个哨兵上线了... +sentinel sentinel 98a51b9a5cf9964b7d17325c58ef6408dfddef20 172.17.0.6 7380 @ mymaster 172.17.0.2 6379 ... * +sentinel sentinel 6aca27450e49ee7de6b5358ab0e91c64dba62d24 172.17.0.7 7380 @ mymaster 172.17.0.2 6379 后启动的哨兵也知道目前存在哪些哨兵和集群中有哪些机器..... +slave slave 172.17.0.3:6379 172.17.0.3 6379 @ mymaster 172.17.0.2 6379 ... * +slave slave 172.17.0.4:6379 172.17.0.4 6379 @ mymaster 172.17.0.2 6379 ... +sentinel sentinel 98a51b9a5cf9964b7d17325c58ef6408dfddef20 172.17.0.6 7380 @ mymaster 172.17.0.2 6379 ...* +sentinel sentinel 8b36fa769b82d9131f3b33c4ef941864fbe36481 172.17.0.5 7380 @ mymaster 172.17.0.2 6379他们是通过什么通信的呢？答案是发布/订阅，可以登录主订阅消息PSUBSCRIBE *，可以知道哨兵在“聊”些什么，如下，你可以看到哨兵之间在"聊天"，实际上：客户端可以将 Sentinel 看作是一个只提供了订阅功能的 Redis 服务器3) "__sentinel__:hello" 4) "172.17.0.5,7380,8b36fa769b82d9131f3b33c4ef941864fbe36481,0,mymaster,172.17.0.2,6379,0" 1) "pmessage" 2) "*" 3) "__sentinel__:hello" 4) "172.17.0.6,7380,98a51b9a5cf9964b7d17325c58ef6408dfddef20,0,mymaster,172.17.0.2,6379,0" 1) "pmessage" 2) "*" 3) "__sentinel__:hello" 4) "172.17.0.7,7380,6aca27450e49ee7de6b5358ab0e91c64dba62d24,0,mymaster,172.17.0.2,6379,0" 1) "pmessage" 测试现在强行停掉主，然后发现从一直在说主挂了，我连不上... * MASTER <-> REPLICA sync started ... # Error condition on socket for SYNC: Connection refused ... * Connecting to MASTER 172.17.0.2:6379 ... * MASTER <-> REPLICA sync started ... # Error condition on socket for SYNC: Connection refused过了好一会儿（因为网络有延迟，有可能是主网络不通了，所以要等一个时间窗），看一下第一个启动的哨兵在干嘛，它说主挂了，通过投币选择了一个新的主，新的主是172.17.0.3，此时可以测试一下主从数据是否同步... # +odown master mymaster 172.17.0.2 6379 #quorum 2/2 # 哨兵发现主挂了 ... # +new-epoch 1 # 开启新纪元 ... # 省略选举过程 ...# +switch-master mymaster 172.17.0.2 6379 172.17.0.3 6379 # 将主从172.17.0.2切换到172.17.0.3 ...* +slave slave 172.17.0.4:6379 172.17.0.4 6379 @ mymaster 172.17.0.3 6379 ...* +slave slave 172.17.0.2:6379 172.17.0.2 6379 @ mymaster 172.17.0.3 6379此时再次启动原先的主172.17.0.2，过了一会儿，可以看到某个哨兵的输出，原来的主变成了从机了！此时新的主依旧是172.17.0.3+convert-to-slave slave 172.17.0.2:6379 172.17.0.2 6379 @ mymaster 172.17.0.3 6379看一下sentinel的配置文件：cat /etc/redis/sentinel7380.conf，发现也发生了变化port 7380 sentinel myid 8b36fa769b82d9131f3b33c4ef941864fbe36481 # Generated by CONFIG REWRITE dir "/root/soft/redis-5.0.5/utils" protected-mode no sentinel deny-scripts-reconfig yes sentinel monitor mymaster 172.17.0.3 6379 2 # 新的主，被重写了 sentinel config-epoch mymaster 1 sentinel leader-epoch mymaster 1 sentinel known-replica mymaster 172.17.0.2 6379 sentinel known-replica mymaster 172.17.0.4 6379 sentinel known-sentinel mymaster 172.17.0.6 7380 98a51b9a5cf9964b7d17325c58ef6408dfddef20 sentinel known-sentinel mymaster 172.17.0.7 7380 6aca27450e49ee7de6b5358ab0e91c64dba62d24 sentinel current-epoch 1 总结：哨兵可以进行故障转移，如果主宕机了，会投票选出一个新的主，然后原来的主上线后自动变成从机，保证服务的高可用代理现在主从集群+哨兵都已经搭建好了，对外可以提供高可用的服务了，但是对于客户端来说要连接集群的那台服务器呢？写的请求是要连接的主的，读请求可以连接到主或者从，这个逻辑要在客户端来控制吗？读请求负载均衡到那台redis服务器，这个负载均衡的逻辑还是要客户端来实现吗？我们发现，虽然Redis集群对外可以提供高可用服务了，但是却增加了客户端的连接成本，作为客户端来讲，我希望Redis集群对外是透明的，只需要一个IP和端口就可以连接整个集群，那么有没有解决方案呢？有的，那就是Redis代理！当然Redis代理有很多，下面是各个Redis代理的比较目前市面上主流的代理包含：predixy、twemproxy、codis、redis-cerberus四款，这四款各有各的优势，我们逐个对比进行对比分析。特性predixytwemproxycodisredis-cerberus高可用Redis Sentinel或Redis Cluster一致性哈希Redis SentinelRedis Cluster可扩展Key哈希分布或Redis ClusterKey哈希分布Key哈希分布Redis Cluster开发语言C++CGOC++多线程是否是是事务Redis Sentinel模式单Redis组下支持不支持不支持不支持BLPOP/BRPOP/BLPOPRPUSH支持不支持不支持支持Pub/Sub支持不支持不支持支持Script支持load不支持不支持不支持Scan支持不支持不支持不支持Select DB支持不支持支持Redis Cluster只有一个DBAuth支持定义多个密码，给予不同读写及管理权限和Key访问空间不支持同redis不支持读从节点支持，可定义丰富规则读指定的从节点不支持支持，简单规则支持，简单规则多机房支持支持，可定义丰富规则调度流量不支持有限支持有限支持统计信息丰富丰富丰富简单在功能的对比上，predixy相比另外三款代理更为全面，基本可以完全适用原生redis的使用场景。在性能上，predixy在各轮测试中都以较大优势领先，参考：https://github.com/joyieldInc/predixy/wiki/Benchmark对各代理的总结如下：predixy：功能全面，既可以使用单个主从redis，也可使用Redis Cluster；性能优异。twemproxy：高可用依赖一致性哈希，仅在缓存场景下适用，不适用存储使用；性能中等。codis：适用redis集群使用；性能一般。cerberus：适用使用Redis Cluster；在数据量较小且pipeline使用情况下性能尚可，否则性能较差。因此，我们选择predixy作为我们Redis集群的代理，predixy项目地址：https://github.com/joyieldInc/predixy环境准备docker run -dit --name p1 --privileged centos /usr/sbin/init它的ip是172.17.0.8编译安装根据README进行编译安装，编译的过程如果报错就根据提示安装相应的编译环境，比如make[1]: g++: Command not found那就安装c++编译器 gcc-c++注意libstdc++-static yum可能安装不上，如果是CnetOS8则可以用如下命令安装# 参考https://centos.pkgs.org/8/centos-powertools-x86_64/libstdc++-static-8.3.1-5.1.el8.x86_64.rpm.html dnf --enablerepo=PowerTools install libstdc++-static编译安装脚本# 安装编译环境，各种工具等，注意libstdc++-static可能安装不上 yum install -y git wget gcc gcc-c++ libstdc++-static make telnet # 创建文件夹 cd ~ && mkdir soft && cd soft # clone代码到本地 git clone https://github.com/joyieldInc/predixy.git # 编译安装 cd predixy && make # 拷贝 mkdir -p /opt/predixy && cp src/predixy /opt/predixy # 添加环境变量 cat >> /etc/profile <<EOF export PATH=$PATH:/opt/predixy EOF echo 'source /etc/profile' >> ~/.bashrc && source ~/.bashrc # 帮助命令 predixy -h配置代理# 拷贝一份配置文件 mkdir -p /etc/predixy && cp -r ~/soft/predixy/conf /etc/predixypredixy的配置类似redis, 具体配置项的含义在配置文件里有详细解释，请参考下列配置文件：predixy.conf：整体配置文件，会引用下面的配置文件cluster.conf：用于Redis Cluster时，配置后端redis信息sentinel.conf：用于Redis Sentinel时，配置后端redis信息auth.conf：访问权限控制配置，可以定义多个验证密码，可每个密码指定读、写、管理权限，以及定义可访问的健空间dc.conf：多数据中心支持，可以定义读写分离规则，读流量权重分配latency.conf： 延迟监控规则定义，可以指定需要监控的命令以及延时时间间隔提供这么多配置文件实际上是按功能分开了，所有配置都可以写到一个文件里，也可以写到多个文件里然后在主配置文件里引用进来。具体配置可以参考官方文档：https://github.com/joyieldInc/predixy/blob/master/README_CN.md开启日志、引入sentinel.conf配置文件#predixy 默认运行在7617端口 # 开启日志 sed -i 's|# Log ./predixy.log|Log ./predixy.log|g' /etc/predixy/conf/predixy.conf # 引入sentinel.conf sed -i 's|# Include sentinel.conf|Include sentinel.conf|g' /etc/predixy/conf/predixy.conf # 注释测试 try.conf sed -i 's|Include try.conf|# Include try.conf|g' /etc/predixy/conf/predixy.conf因为我们的集群就是主从集群+sentinel监控，所以主要配置sentinel.conf# 备份一下 cp /etc/predixy/conf/sentinel.conf /etc/predixy/conf/sentinel.conf.bak # 写入配置 cat > /etc/predixy/conf/sentinel.conf <<EOF SentinelServerPool { # 没有密码 # Password # 主从都是全量的，并且只有一组集群，不涉及到hash分片 # Hash crc16 # HashTag "{}" # Distribution modula Databases 16 MasterReadPriority 0 # master禁止读 StaticSlaveReadPriority 50 DynamicSlaveReadPriority 50 RefreshInterval 1 ServerTimeout 1 ServerFailureLimit 10 ServerRetryTimeout 1 KeepAlive 120 Sentinels { # 配置哨兵的ip + 172.17.0.5:7380 + 172.17.0.6:7380 + 172.17.0.7:7380 } # Group后面的名字必须和哨兵配置的名字相同，比如上面配置的是mymaster，这里也应该是mymaster Group mymaster { # 配置redis主从集群IP，配置了就是静态，对应StaticSlaveReadPriority # 不配置就是动态，对用DynamicSlaveReadPriority + 172.17.0.2:6379 + 172.17.0.3:6379 + 172.17.0.4:6379 } } EOF一定要注意Group后面的参数一定是哨兵中定义的名字，否则会报错(error) ERR no server avaliable，Group里面配置了的IP读优先级对应为StaticSlaveReadPriority，否则通过哨兵发现的节点的优先级对应DynamicSlaveReadPriority如果把主从都配置成静态的，StaticSlaveReadPriority设置成0，而DynamicSlaveReadPriority设置成100猜猜会发生什么情况？配置的静态但是没有优先级？那么也会报错(error) ERR no server avaliable参数说明Password: 指定连接redis实例默认的密码，不指定的情况下表示redis不需要密码Databases: 指定redis db数量，不指定的情况下为1Hash: 指定对key算哈希的方法，当前只支持atol和crc16HashTag: 指定哈希标签，不指定的话为{}Distribution: 指定分布key的方法，当前只支持modula和randomMasterReadPriority: 读写分离功能，从redis master节点执行读请求的优先级，为0则禁止读redis master，不指定的话为50StaticSlaveReadPriority: 读写分离功能，从静态redis slave节点执行读请求的优先级，所谓静态节点，是指在本配置文件中显示列出的redis节点，不指定的话为0DynamicSlaveReadPolicy: 功能见上，所谓动态节点是指在本配置文件中没有列出，但是通过redis sentinel动态发现的节点，不指定的话为0RefreshInterval: predixy会周期性的请求redis sentinel以获取最新的集群信息，该参数以秒为单位指定刷新周期，不指定的话为1秒ServerTimeout: 请求在predixy中最长的处理/等待时间，如果超过该时间redis还没有响应的话，那么predixy会关闭同redis的连接，并给客户端一个错误响应，对于blpop这种阻塞式命令，该选项不起作用，为0则禁止此功能，即如果redis不返回就一直等待，不指定的话为0ServerFailureLimit: 一个redis实例出现多少次才错误以后将其标记为失效，不指定的话为10ServerRetryTimeout: 一个redis实例失效后多久后去检查其是否恢复正常，不指定的话为1秒KeepAlive: predixy与redis的连接tcp keepalive时间，为0则禁止此功能，不指定的话为0Sentinels: 里面定义redis sentinel实例的地址Group: 定义一个redis组，Group的名字应该和redis sentinel里面的名字一致，Group里可以显示列出redis的地址，列出的话就是上面提到的静态节点启动并测试启动代理# 启动 predixy /etc/predixy/conf/predixy.conf & # 查看日志 tail -f /etc/predixy/conf/predixy.log 日志输出如下，说明已经启动... N Proxy.cpp:112 predixy listen in 0.0.0.0:7617 ... N Proxy.cpp:143 predixy running with Name:PredixyExample Workers:1 ... N Handler.cpp:456 h 0 create connection pool for server 172.17.0.5:7380 ...随便找一台有redis-cli的机器测试一下# 连接代理 redis-cli -h 172.17.0.8 -p 7617 # 设置值 set k1 aaa # 获取值 get k1set k1 aaa的时候通过日志可以看到predixy是懒加载的，set的时候才去创建主的连接池...ConnectConnectionPool.cpp:42 h 0 create server connection 172.17.0.3:6379 10get k1的时候观察日志输出，当第一次get操作的时候创建了从172.17.0.2的连接池...ConnectConnectionPool.cpp:42 h 0 create server connection 172.17.0.2:6379 11后续的某次get操作会创建从172.17.0.4的连接池...ConnectConnectionPool.cpp:42 h 0 create server connection 172.17.0.4:6379 12可见写请求都是走主。读请求都是走从并且自带负载均衡！整体的架构如下图代理单点故障聪明的你肯定想到了，这predixy就只有一个，那么当他挂了怎么办？这里其实存在一个单点故障，那么可以用keepalived技术做一个高可用，关于keepalived之前的文章已有项目描述，此处不再赘述，感兴趣的同学可以翻翻之前的文章或者谷歌一下水平扩容如果我们有多组Master/Slave集群，就可以进行水平扩容，将业务主键通过Hash算法后分布在不同的集群下，那么predixy就可以这样配置cat > /etc/predixy/conf/sentinel.conf <<EOF SentinelServerPool { # 没有密码 # Password # 主从都是全量的，有2组集群，可以Hash进行数据分片 Hash crc16 HashTag "{}" Distribution modula Databases 16 MasterReadPriority 0 # master禁止读 StaticSlaveReadPriority 50 DynamicSlaveReadPriority 50 RefreshInterval 1 ServerTimeout 1 ServerFailureLimit 10 ServerRetryTimeout 1 KeepAlive 120 Sentinels { # 配置哨兵的ip + 172.17.0.5:7380 + 172.17.0.6:7380 + 172.17.0.7:7380 } # Group后面的名字必须和哨兵配置的名字相同，比如上面配置的是mymaster，这里也应该是mymaster Group mymaster { # 配置redis主从集群IP，配置了就是静态，对应StaticSlaveReadPriority # 不配置就是动态，对用DynamicSlaveReadPriority + 172.17.0.2:6379 + 172.17.0.3:6379 + 172.17.0.4:6379 } Group mymaster-slave { + 172.17.0.20:6379 + 172.17.0.30:6379 + 172.17.0.40:6379 } } EOF以上配置有mymaster和mymaster-slave两组Master/Slave集群，通过crc16算法计算业务主键的hash值，并且通过modula也就是取模的办法得出应该在哪个集群里面去读写。这样看似不错，但是有一个问题，如果再增加一个集群，那么大部分缓存取模的值就不一样了，意为着大量缓存都需要重建！这当然是不利于水平扩展的，架构图如下所以怎么解决这个问题呢？答案是一致性Hash环算法和Hash槽，即Hash Slot，Redis自带的Cluster集群模式自带Hash Slot算法，我们下篇文章将会讲Redis Cluster的搭建。参考https://github.com/joyieldInc/predixy

5张图讲明白JDK1.7下的HashMap死循环（原理+实战） 5张图讲明白JDK1.7下的HashMap死循环（原理+实战）情景再现网络上很多文章说HashMap死循环都是理论分析，其一是没有自己实验过，甚至给出的实验程序都是错误的，不能再现死循环的BUG，其二是给出的示意图不够详细，很多细节忽略了让人难以理解，本文在前人总结的基础上摸索出了实现死循环的方法，记录成文。死循环原因我们首先思考一下在什么样的情况下HashMap会死循环，死循环的原因不外乎是在多线程的同时扩容，在JDK 1.7的HashMap中，当hash冲突时，采用头插法拉链表，所谓头插法，即在每次都在链表头部（即桶中）插入最后添加的数据当触发扩容阈值准备扩容的时候，会循环旧桶中的每个元素，重新计算hash值然后再次分配到新的桶中，这个过程如果产生hash冲突，那也会采用头插法拉链表，循环的时候是从头到尾，而头插法插入的时候，尾巴上的元素反而会变成头元素，相当于逆序了当一个线程A在扩容还未完成的时候，不巧的是失去了CPU时间片，不能获得执行机会，另外一个线程B抢先扩容完毕了，并且链表变成了逆序，而线程A还在按照顺序操作，造成指针混乱，于是出现死循环，当然这么说过于笼统也说不清楚，后文讲解原理的时候会详细到每个步骤！实验环境IDEA + JDK 1.7，自从Oracle改版以后需要登录才能下载，而且链接不是很好找，这里推荐一个下载站点：http://java.sousou88.com/spec/oraclejdk.html实验思路既然要多线程才会实现，那么可以利用IDEA的多线程DEBUG操作将两个线程同时停在扩容操作处，然后让其中一个线程先扩容完，另外一个线程再次扩容造成死循环程序代码 import java.io.IOException; import java.util.HashMap; /** * 测试在JDK 1.7中 HashMap出现死循环 */ public class Main { /** * 这个map 桶的长度为2，当元素个数达到 2 * 1.5 = 3 的时候才会触发扩容 */ private static HashMap<Integer,String> map = new HashMap<Integer,String>(2,1.5f); public static void main(String[] args) throws IOException, InterruptedException { map.put(5,"5"); map.put(7,"7"); map.put(3,"3"); System.out.println("此时元素已经达到3了，再往里面添加就会产生扩容操作：" + map); new Thread("T1") { public void run() { map.put(11, "11"); System.out.println(Thread.currentThread().getName() + "扩容完毕 " ); }; }.start(); new Thread("T2") { public void run() { map.put(15, "15"); System.out.println(Thread.currentThread().getName() + "扩容完毕 " + map); }; }.start(); Thread.sleep(10000);//时间根据debug时间调整 //死循环后打印直接OOM，思考一下为什么？ //因为打印的时候回调用toString回遍历链表，但此时链表已经成环状了 //那么就会无限拼接字符串 // System.out.println(map); System.out.println(map.get(5)); System.out.println(map.get(7)); System.out.println(map.get(3)); System.out.println(map.get(11)); System.out.println(map.get(15)); } } 注意环境是JDK 1.7，首先分析一下程序，我们创建了一个桶长度为2，负载因子为1.5的HashMap，也就是说这个Map在实际元素数量达到2 * 1.5 = 3之后，也就是放入第4个元素的时候会触发扩容，最开始我们放入了3个元素5、7、3，那么根据hash布局如下此时我们开启线程T1和T2同时再往map里面put值，超过3个了，要扩容了！如果两个线程在并行，那么这两个线程都会引发扩容操作！此时我们可以通过DEBUG手动做一些手脚，最后再分别取出我们put的值看看是什么，下面就是我们做的手脚！操作步骤定义断点在HashMap源码第589行transfer方法第一句话打一个条件断点条件如下，意为当线程是T1或者T2的时候停住，其他放行Thread.currentThread().getName().equals("T1")||Thread.currentThread().getName().equals("T2")需要注意的一点是Suspend选择Thread，表示是多线程DEBUG，多个线程会同时停住然后以DEBUG的模式启动程序main方法线程T1操作不用意外将会在此处停住，左下角可以看到T1和T2同时都在DEBUG，也就是同时在扩容！然后手动放行T1到图示位置，也就是在这句代码之下！Entry<K,V> next = e.next;很重要！否则可能造成不了死循环，因为这句话给e和next两个变量赋值了，才会造成后面的指针混乱！线程T2操作T2的操作就比较简单了，选择T2线程，直接放行即可，让T2先进行扩容！线程T2放行以后会再次停留在T1线程DEBUG处，因为T1线程我们只做了赋值还没进行后续扩容操作，直接放行T1即可！看结果如果DEBUG时间操作太长main线程已经执行完毕，可以将main线程里面的时间改长一点 Thread.sleep(10000);//时间根据debug时间调整 不出意外，会输出下面的值，我们put的5、7、3、11、15结果只有7、3、11，5和15已经不见了！并且程序卡住不动了，说明已经死循环了，再看看电脑CPU占用是否不正常！此时元素已经达到3了，再往里面添加就会产生扩容操作：{3=3, 7=7, 5=5} T2扩容完毕 {5=5, 15=15, 7=7, 3=3} T1扩容完毕 null 7 3 11 原理分析大概原理之前已经提到过：多个线程同时对HashMap做扩容操作可能会导致一个线程以另一个线程扩容完的逆序链表进行顺序操作，导致循环链表的产生，那么我们来看看具体是怎么产生的基本条件构造private static HashMap<Integer,String> map = new HashMap<Integer,String>(2,1.5f); ... map.put(5,"5"); map.put(7,"7"); map.put(3,"3");这几句代码就是我们的基本条件构造，桶长度为2，扩容阈值为 2*1.5=3，5、3、7设计的很巧妙刚好hash冲突并在一条链表上，在put一个值就会产生扩容操作，基本内存布局如下那么为什么要这么构造？桶长度为2那么下次扩容肯定是4，3%4 = 7%4，而5%4=1不再一个桶上了，所以扩容后至少会有3和7两个元素还在一个链条上，这样才有产生死循环的机会T2扩容完毕T2扩容完后布局如下（上面一个图），而此时T1因为只执行了Entry<K,V> next = e.next（初始化了e和next指针），并不知道T2已经扩容完毕了，所以还在按照只有两个桶的布局来扩容，但是T1读取数据的时候又是按照T2已经扩容完毕的数据进行读取的！初始化的两个指针值为（按照只有两个桶的时候初始化的指针，因为此时T2还未扩容）e：3 next：7T1扩容第一轮循环第一轮循环毫无疑问将3放到了桶上，然后循环体最后一句话 e = next;将next赋值给e让e指向了7T1扩容第二轮循环上轮循环将e指向了7，本次循环的第一句话 Entry<K,V> next = e.next;，e.next是谁？这是关键，e.next已经被T2扩容的时候改成了3！但是3我们上面第一轮循环已经处理过了啊！这里势必会出问题，不过将在下轮循环中出问题，本轮循环e为7，直接将7插入到头部（桶中），那么这里就有一个7 > 3的引用T1扩容第三轮循环问题就来了，上轮循环将e指向了3，3后面是null（T2扩容处理的），但是3我们已经在第一轮处理过了，并且在第二轮有个7 > 3的引用，本轮循环又会将3放到头部（桶中），3.next将会指向7产生一个3 > 7的引用，这不就循环引用产生循环链表了？问题就这么产生了！最后别忘了T1的原本意图是要将11放入map，最后将11放入头部（桶中），所以最后的内存布局如下：当我们分别取拿5、7、3、11、15的值的时候，对比上面布局就应该知道能不能拿到System.out.println(map.get(5));//没有了，拿不到 System.out.println(map.get(7));//可以拿到 System.out.println(map.get(3));//可以拿到 System.out.println(map.get(11));//可以拿到 System.out.println(map.get(15));//死循环！！！和我们的输出结果一模一样null 7 3 11 ...死循环中总结希望通过上面的图示你能理解JDK1.7中死循环产生的原因，那么如何避免呢？答案当然是采用线程安全的ConcurrentHashMap，而且HashMap的实现在JDK1.8中有很多优化，并且也采用尾插法解决了死循环问题

深入浅出keepalived+nginx实现网关主备高可用 深入浅出keepalived+nginx实现网关主备高可用背景当用Nginx作网关的时候，如果网关宕机了，整个服务将会变得不可用，那么如何保证网关的高可用呢？我们可以用Keepalived来做主备，实现网关的高可用，主机宕机了，备机选举出一个来IP自动漂移顶上去，主机恢复了，IP再飘回主机，备机自动下线！思考一下，代理可以实现高可用不呢？在两台网关前面加一台代理，轮询将请求分别打给两台网关，当一台网关宕机了，代理就不会往这台网关上分发流量，这样也能实现高可用，但是代理自身也只有一个，自己宕机了那么整个服务就不可用，也就是代理自己存在单点故障，那么在代理前面在加一层代理？会发现无论前置多少层代理，无论怎么无限套娃总会存在代理自身单点故障，所以代理是不能实现自身的高可用的，它只能实现自己后层的服务的高可用。keepalived原理研究东西首先研究官网，官网说：keepalive是一个用c语言编写的路由软件。这个项目的主要目标是为Linux系统和基于Linux的基础设施提供简单、健壮的负载平衡和高可用性工具，通过VRRP协议实现高可用性。VRRP是路由器故障切换的基础...https://www.keepalived.org/那么我们可以提炼出关键词：VRRP协议，那么这个协议是个啥？我们首先思考一下：网关要实现主备，那么IP地址怎么设？不能主备都设置同一个IP吧，这样就冲突了，所以一定得有一个虚拟的IP用于接入网络，主备各自得有一个自己的IP用于主备自己的通信吧，不然备机怎么知道主宕机了，要么是备机不断的去请求主机看主机有没有返回数据包，一定时间没有返回就说明主机宕机了，要么是主机主动给备机发数据包，备机检测到一定时间内没有返回数据包则说明主宕机了，那么这个通知的过程得有一个通信协议吧，所谓协议，就是主备之间通信的方式，它定义了通信的内容，所以整个过程我们可以提炼出下面的概念VIP：虚拟IP（Virtual IP），即接入网络用于客户端访问的那个IPMASTER：主机，任何时候主机只有一台BACKUP：备机/备机组，备机可能不止一台VRRP：这就是我们所说的协议，VRRP全称为Virtual Router Redundancy Protocol，即虚拟路由器冗余协议所以直接过过一遍VRRP协议内容VRRP协议VRRP协议有两个版本VRRPv2报文结构VRRPv3报文结构报文字段含义 VRRPv2VRRPv3VersionVRRP协议版本号，取值为2。VRRP协议版本号，取值为3。TypeVRRP通告报文的类型，取值为1，表示Advertisement。VRRP通告报文的类型，取值为1，表示Advertisement。Virtual Rtr ID（VRID）虚拟路由器ID，取值范围是1～255。虚拟路由器ID，取值范围是1～255。PriorityMaster设备在备份组中的优先级，取值范围是0～255。0表示设备停止参与VRRP备份组，用来使备份设备尽快成为Master设备，而不必等到计时器超时；255则保留给IP地址拥有者。缺省值是100。Master设备在备份组中的优先级，取值范围是0～255。0表示设备停止参与VRRP备份组，用来使备份设备尽快成为Master设备，而不必等到计时器超时；255则保留给IP地址拥有者。缺省值是100。Count IP Addrs/Count IPvX Addr备份组中虚拟IPv4地址的个数。备份组中虚拟IPv4或虚拟IPv6地址的个数。Auth TypeVRRP报文的认证类型。协议中指定了3种类型：0：Non Authentication，表示无认证。1：Simple Text Password，表示明文认证方式。2：IP Authentication Header，表示MD5认证方式。-Adver Int/Max Adver IntVRRP通告报文的发送时间间隔，单位是秒，缺省值为1秒。VRRP通告报文的发送时间间隔，单位是厘秒，缺省值为100厘秒（1秒）。Checksum16位校验和，用于检测VRRP报文中的数据破坏情况。16位校验和，用于检测VRRP报文中的数据破坏情况。IP Address/IPvX Address(es)VRRP备份组的虚拟IPv4地址，所包含的地址数定义在Count IP Addrs字段。VRRP备份组的虚拟IPv4地址或者虚拟IPv6地址，所包含的地址数定义在Count IPvX Addrs字段。Authentication DataVRRP报文的认证字。目前只有明文认证和MD5认证才用到该部分，对于其它认证方式，一律填0。-rsvd-VRRP报文的保留字段，必须设置为0。状态机VRRP用状态机来完成了主备之间的切换，所谓状态机，一言以蔽之状态机是有限状态自动机的简称，是现实事物运行规则抽象而成的一个数学模型注意两个特征：有限和自动，有限即状态是有限的，每个状态带有一系列的动作，当处于某状态下，这些动作会自动执行，在某个事件下，会触发一个状态到另外一个状态的切换，比如自动售卖机可能有三个状态未付款状态：什么也不做付款中状态：检查是否足额付款已足额付款状态：弹出汽水；余额清零；销售额累加等等那么此例中状态是有限的，跟随状态的动作会在该状态下自动执行，投币的动作会触发状态的切换，因此引出状态机的四个概念State：状态，一个状态机至少要包含两个状态Action：动作，事件发生以后要执行动作Event：事件，变换到某个状态的触发条件Transition：变换，也就是从一个状态变化为另一个状态VRRP协议定义了一台机器的三种状态：初始状态（Initialize）活动状态（Master）：只有处于Master状态的设备才可以转发那些发送到虚拟IP地址的报文备份状态（Backup）以下是他们的切换关系那么在不同状态下的动作又是什么呢？Initialize该状态为VRRP不可用状态，在此状态时设备不会对VRRP报文做任何处理。通常刚配置VRRP时或设备检测到故障时会进Initialize状态。收到接口Up的消息后，如果设备的优先级为255，则直接成为Master设备；如果设备的优先级小于255，则会先切换至Backup状态。Masster定时（Advertisement Interval）发送VRRP通告报文。以虚拟MAC地址响应对虚拟IP地址的ARP请求。转发目的MAC地址为虚拟MAC地址的IP报文。如果它是这个虚拟IP地址的拥有者，则接收目的IP地址为这个虚拟IP地址的IP报文。否则，丢弃这个IP报文。如果收到比自己优先级大的报文，立即成为Backup。如果收到与自己优先级相等的VRRP报文且本地接口IP地址小于对端接口IP，立即成为Backup。Backup接收Master设备发送的VRRP通告报文，判断Master设备的状态是否正常。对虚拟IP地址的ARP请求，不做响应。丢弃目的IP地址为虚拟IP地址的IP报文。如果收到优先级和自己相同或者比自己大的报文，则重置Master_Down_Interval定时器，不进一步比较IP地址。Master_Down_Interval定时器：Backup设备在该定时器超时后仍未收到通告报文，则会转换为Master状态。计算公式如下：Master_Down_Interval=(3*Advertisement_Interval) + Skew_time。其中，Skew_Time=(256–Priority)/256。如果收到比自己优先级小的报文且该报文优先级是0时，定时器时间设置为Skew_time（偏移时间），如果该报文优先级不是0，丢弃报文，立刻成为Master。实战环境准备利用docker创建两个CentOS容器ngt1与ngt2，用docker的原因是降低实战的成本与复杂度docker run -dit --name ngt1 --privileged centos /usr/sbin/init docker run -dit --name ngt2 --privileged centos /usr/sbin/initngt1的ip是 172.17.0.2ngt2的ip是 172.17.0.3nginx安装上面的两个容器ngt1和ngt2都需要安装编译安装脚本如下# 定义路径、定义版本 NGINX_HOME=/usr/local/nginx NGINX_VERSION=1.16.0 mkdir -p $NGINX_HOME O_PATH=`pwd` # 安装依赖wget 下载 gcc gcc-c++ 为编译环境 pcre pcre-devel使nginx支持rewrite模块 # openssl openssl-devel 使nginx支持ssl zlib zlib-devel gd gd-devel 使nginx支持压缩 gd用于支持图像压缩 yum install -y wget net-tools make cmake gcc gcc-c++ pcre pcre-devel openssl openssl-devel zlib zlib-devel gd gd-devel # 创建用户 useradd -s /sbin/nologin nginx # 下载并解压 cd $NGINX_HOME && wget wget http://nginx.org/download/nginx-$NGINX_VERSION.tar.gz tar -zxf nginx-$NGINX_VERSION.tar.gz # 编译安装 cd nginx-$NGINX_VERSION ./configure --prefix=$NGINX_HOME \ --user=nginx \ --group=nginx \ --with-pcre \ --with-http_ssl_module \ --with-http_v2_module \ --with-http_realip_module \ --with-http_addition_module \ --with-http_sub_module \ --with-http_dav_module \ --with-http_flv_module \ --with-http_mp4_module \ --with-http_gunzip_module \ --with-http_gzip_static_module \ --with-http_random_index_module \ --with-http_secure_link_module \ --with-http_stub_status_module \ --with-http_auth_request_module \ --with-http_image_filter_module \ --with-http_slice_module \ --with-mail \ --with-threads \ --with-file-aio \ --with-stream \ --with-mail_ssl_module \ --with-stream_ssl_module make && make install # 软链 ln -s $NGINX_HOME/sbin/nginx /sbin/ # 添加环境变量 echo "export NGINX_HOME=$NGINX_HOME" >> ~/.bashrc . ~/.bashrc cd $O_PATH # 查看版本 nginx -V配置文件位置：cat /usr/local/nginx/conf/nginx.conf修改HTML将index.html修改成容器ip方便观察# 备份 \cp /usr/local/nginx/html/index.html /usr/local/nginx/html/index.html.bak # 设置IP echo `ifconfig $name | grep "inet.*broadcast.*" | cut -d' ' -f10` > /usr/local/nginx/html/index.html启动nginx 检测是否启动成功curl 127.0.0.1如果回显刚才设置的ip说明成功nginx编译参数解释--with-cc-opt='-g -O2 -fPIE -fstack-protector' # 设置额外的参数将被添加到CFLAGS变量。（FreeBSD或者ubuntu使用） --param=ssp-buffer-size=4 -Wformat -Werror=format-security -D_FORTIFY_SOURCE=2' --with-ld-opt='-Wl,-Bsymbolic-functions -fPIE -pie -Wl,-z,relro -Wl,-z,now' --prefix=/usr/share/nginx # 指向安装目录 --conf-path=/etc/nginx/nginx.conf # 指定配置文件 --http-log-path=/var/log/nginx/access.log # 指定访问日志 --error-log-path=/var/log/nginx/error.log # 指定错误日志 --lock-path=/var/lock/nginx.lock # 指定lock文件 --pid-path=/run/nginx.pid # 指定pid文件 --http-client-body-temp-path=/var/lib/nginx/body # 设定http客户端请求临时文件路径 --http-fastcgi-temp-path=/var/lib/nginx/fastcgi # 设定http fastcgi临时文件路径 --http-proxy-temp-path=/var/lib/nginx/proxy # 设定http代理临时文件路径 --http-scgi-temp-path=/var/lib/nginx/scgi # 设定http scgi临时文件路径 --http-uwsgi-temp-path=/var/lib/nginx/uwsgi # 设定http uwsgi临时文件路径 --with-debug # 启用debug日志 --with-pcre-jit # 编译PCRE包含“just-in-time compilation” --with-ipv6 # 启用ipv6支持 --with-http_ssl_module # 启用ssl支持 --with-http_stub_status_module # 获取nginx自上次启动以来的状态 --with-http_realip_module # 允许从请求标头更改客户端的IP地址值，默认为关 --with-http_auth_request_module # 实现基于一个子请求的结果的客户端授权。如果该子请求返回的2xx响应代码，所述接入是允许的。如果它返回401或403中，访问被拒绝与相应的错误代码。由子请求返回的任何其他响应代码被认为是一个错误。 --with-http_addition_module # 作为一个输出过滤器，支持不完全缓冲，分部分响应请求 --with-http_dav_module # 增加PUT,DELETE,MKCOL：创建集合,COPY和MOVE方法 默认关闭，需编译开启 --with-http_geoip_module # 使用预编译的MaxMind数据库解析客户端IP地址，得到变量值 --with-http_gunzip_module # 它为不支持“gzip”编码方法的客户端解压具有“Content-Encoding: gzip”头的响应。 --with-http_gzip_static_module # 在线实时压缩输出数据流 --with-http_image_filter_module # 传输JPEG/GIF/PNG 图片的一个过滤器）（默认为不启用。gd库要用到） --with-http_spdy_module # SPDY可以缩短网页的加载时间 --with-http_sub_module # 允许用一些其他文本替换nginx响应中的一些文本 --with-http_xslt_module # 过滤转换XML请求 --with-mail # 启用POP3/IMAP4/SMTP代理模块支持 --with-mail_ssl_module # 启用ngx_mail_ssl_module支持启用外部模块支持nginx常用命令nginx -c /path/to/nginx.conf # 以特定目录下的配置文件启动nginx: nginx -s reload # 修改配置后重新加载生效 nginx -s reopen # 重新打开日志文件 nginx -s stop # 快速停止nginx nginx -s quit # 完整有序的停止nginx nginx -t # 测试当前配置文件是否正确 nginx -t -c /path/to/nginx.conf # 测试特定的nginx配置文件是否正确keepalived安装编译安装脚本如下：# 定义路径、定义版本 KEEPALIVED_HOME=/usr/local/keepalived KEEPALIVED_VERSION=2.1.5 mkdir -p $KEEPALIVED_HOME O_PATH=`pwd` # 安装依赖 yum install -y wget make cmake gcc gcc-c++ openssl-devel net-tools # 下载并解压 cd $KEEPALIVED_HOME && \ wget https://www.keepalived.org/software/keepalived-$KEEPALIVED_VERSION.tar.gz && \ tar -zxf keepalived-$KEEPALIVED_VERSION.tar.gz # 编译安装 cd keepalived-$KEEPALIVED_VERSION ./configure --prefix=$KEEPALIVED_HOME make && make install # 软连 ln -s $KEEPALIVED_HOME/sbin/keepalived /sbin/ # 拷贝配置文件，默认路径为/etc/keepalived mkdir /etc/keepalived \cp -rf $KEEPALIVED_HOME/etc/keepalived/keepalived.conf /etc/keepalived/ # 添加环境变量 echo "export KEEPALIVED_HOME=$KEEPALIVED_HOME" >> ~/.bashrc . ~/.bashrc cd $O_PATH配置检测脚本思考一下，我们的目的是要实现nginx的高可用，那么一定要保证keepalived在运行的时候nginx也在运行，否在备机成为Master但是nginx都没运行呢那服务也是不可用的，简而言之，要用脚本来保证keepalived和nginx处于同时运行的状态，要么同时运行，要么同时不运行脚本如下：在所有主备上都执行下面脚本# 定义检测脚本文件 CHECK_SCRIPT=/etc/keepalived/nginx_check.sh # 确保目录存在 mkdir -p $CHECK_SCRIPT rm -rf $CHECK_SCRIPT yum install -y psmisc # 生成检测脚本 cat > $CHECK_SCRIPT <<EOF #!/bin/bash A=\`ps -C nginx --no-header | wc -l\` if [ \$A -eq 0 ];then nginx sleep 2 if [ \`ps -C nginx --no-header |wc -l\` -eq 0 ];then # 如果没有killall 请安装 yum install -y psmisc killall keepalived fi fi EOF # 添加权限 chmod +x $CHECK_SCRIPT keepalived配置在所有机器上备份一下原始配置文件# 备份keepalived原配置文件 cp /etc/keepalived/keepalived.conf /etc/keepalived/keepalived.conf.bakMaster配置将下面内容写入配置文件/etc/keepalived/keepalived.conf中mcast_src_ip：本机 IP 地址 即ngt1的自己的IP地址，注意本地ip地址在docker中可以在容器内执行ifconfig查看，也可在容器外执行docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' 容器名字查看virtual_ipaddress：虚拟 IP 池, 集群内所有节点设置必须一样，虚拟IP即客户端访问时用的IP，比如172.17.0.100/16 dev eth0 label eth0:vip表示集群对外提供的IP为172.17.0.100，即客户端访问此IP地址，绑定在eth0网卡上，标签为eth0:vip，即eth0设备上的虚拟IP! Configuration File for keepalived global_defs { ## keepalived 自带的邮件提醒需要开启 sendmail 服务。建议用独立的监控或第三方 SMTP ## 标识本节点的字条串，通常为 hostname router_id NGT1 } ## keepalived 会定时执行脚本并对脚本执行的结果进行分析，动态调整 vrrp_instance 的优先级。如果脚本执行结果为 0，并且 weight 配置的值大于 0，则优先级相应的增加。如果脚本执行结果非 0，并且 weight配置的值小于 0，则优先级相应的减少。其他情况，维持原本配置的优先级，即配置文件中 priority 对应的值。 vrrp_script chk_nginx { ## 检测 nginx 状态的脚本路径，即上面的/etc/keepalived/nginx_check.sh script "/etc/keepalived/nginx_check.sh" ## 检测时间间隔 interval 4 ## 如果条件成立，权重-20 weight -20 } ## 定义虚拟路由，VI_1 为虚拟路由的标示符，自己定义名称 vrrp_instance VI_1 { ## 主节点为 MASTER，对应的备份节点为 BACKUP state MASTER ## 绑定虚拟 IP 的网络接口，与本机 IP 地址所在的网络接口相同 interface eth0 ## 虚拟路由的 ID 号，两个节点设置必须一样，可选 IP 最后一段使用，相同的 VRID 为一个组，他将决定多播的 MAC 地址 virtual_router_id 9 ## 本机 IP 地址 即ngt1的自己的IP地址 mcast_src_ip 172.17.0.2 ## 节点优先级，值范围 0-254，MASTER 要比BACKUP高 priority 100 ## 优先级高的设置 nopreempt 解决异常恢复后再次抢占的问题 nopreempt ## 组播信息发送间隔，两个节点设置必须一样，默认 1s advert_int 1 ## 设置验证信息，两个节点必须一致 authentication { auth_type PASS auth_pass 1111 } ## 将 track_script 块加入instance 配置块 track_script { ## 执行 Nginx 监控的服务 chk_nginx } ## 虚拟 IP 池, 集群内所有节点设置必须一样，虚拟IP即客户端访问时用的ip virtual_ipaddress { 172.17.0.100/16 dev eth0 label eth0:vip } } Backup配置参考Master的配置，有几个地方和Master不一样router_id：标识自己，集群内唯一state：备机就配置成BACKUPmcast_src_ip：备机自己的IPpriority：优先级，备机比主机低虚拟IP一定要配置成一样virtual_ipaddress：要和主机配置成一样，表示此集群对外暴露的IP! Configuration File for keepalived global_defs { ## keepalived 自带的邮件提醒需要开启 sendmail 服务。建议用独立的监控或第三方 SMTP ## 标识本节点的字条串，通常为 hostname router_id NGT2 } ## keepalived 会定时执行脚本并对脚本执行的结果进行分析，动态调整 vrrp_instance 的优先级。如果脚本执行结果为 0，并且 weight 配置的值大于 0，则优先级相应的增加。如果脚本执行结果非 0，并且 weight配置的值小于 0，则优先级相应的减少。其他情况，维持原本配置的优先级，即配置文件中 priority 对应的值。 vrrp_script chk_nginx { ## 检测 nginx 状态的脚本路径，即上面的/etc/keepalived/nginx_check.sh script "/etc/keepalived/nginx_check.sh" ## 检测时间间隔 interval 4 ## 如果条件成立，权重-20 weight -20 } ## 定义虚拟路由，VI_1 为虚拟路由的标示符，自己定义名称 vrrp_instance VI_1 { ## 主节点为 MASTER，对应的备份节点为 BACKUP state BACKUP ## 绑定虚拟 IP 的网络接口，与本机 IP 地址所在的网络接口相同 interface eth0 ## 虚拟路由的 ID 号，两个节点设置必须一样，可选 IP 最后一段使用，相同的 VRID 为一个组，他将决定多播的 MAC 地址 virtual_router_id 9 ## 本机 IP 地址 即ngt1的自己的IP地址 mcast_src_ip 172.17.0.3 ## 节点优先级，值范围 0-254，MASTER 要比BACKUP高 priority 80 ## 优先级高的设置 nopreempt 解决异常恢复后再次抢占的问题 nopreempt ## 组播信息发送间隔，两个节点设置必须一样，默认 1s advert_int 1 ## 设置验证信息，两个节点必须一致 authentication { auth_type PASS auth_pass 1111 } ## 将 track_script 块加入instance 配置块 track_script { ## 执行 Nginx 监控的服务 chk_nginx } ## 虚拟 IP 池, 集群内所有节点设置必须一样，虚拟IP即客户端访问时用的ip virtual_ipaddress { 172.17.0.100/16 dev eth0 label eth0:vip } } 启动在所有机器上启动keepalivedkeepalived 查看是否启动ps aux | grep keepalived有如下回显表示启动成功root 9761 0.0 0.0 22492 616 ? Ss 01:36 0:00 keepalived root 9762 0.0 0.1 45480 2352 ? S 01:36 0:00 keepalived在主机和备机上查看VIP，即虚拟IP，命令为：ifconfig，如下回显说明主机上的VIP已经生效了，在备机上同样查看，发现备机并没有VIP，这才是正常的eth0:vip: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500 inet 172.17.0.100 netmask 255.255.0.0 broadcast 0.0.0.0 ether 02:42:ac:11:00:02 txqueuelen 0 (Ethernet)测试在集群外循环访问VIP：发现现在是主机的nginx在对外提供服务while true; do curl 172.17.0.100; sleep 1 ;done 172.17.0.2 172.17.0.2 172.17.0.2 172.17.0.2 ... 现在我们强行将主机上的nginx关闭，在ngt1上执行nginx -s stop，有如下回显172.17.0.2 curl: (7) Failed connect to 172.17.0.100:80; 拒绝连接 curl: (7) Failed connect to 172.17.0.100:80; 拒绝连接 curl: (7) Failed connect to 172.17.0.100:80; 拒绝连接 172.17.0.2 172.17.0.2 172.17.0.2 ... 暂时的停止了服务，稍后又恢复了，这说明是我们的检测脚本执行了，自动恢复nginx的运行现在我们来模拟一下主机宕机，在主机上执行killall keepalived nginx，再观察输出... 172.17.0.2 172.17.0.2 172.17.0.2 172.17.0.3 172.17.0.3 172.17.0.3 ...发现备机自动顶上来了，此时在备机上执行ifconfig是可以看到VIP的，即我们配置的eth0:vip最后我们再次模拟主机恢复，在主机上执行keepalived 启动之，它会根据脚本自动启动nginx，稍等片刻，我们就发现主机自动切换回来了！... 172.17.0.3 172.17.0.3 172.17.0.2 172.17.0.2 172.17.0.2 ...这就是keepalived实现的功能，准确说是VRRP协议能实现的功能，集群内的机器宕机后通过IP漂移（VRRP协议保证）让其他备机顶替上来继续工作，整个过程对客户端是透明的，无感知的！抓包抓包之前再看一下VRRP协议的基本描述：VRRP协议只有一种报文，即主路由器定时向其它成员发送的组播报文。当Master正常工作时，它会每隔一段时间(缺省为1秒)发送一个VRRP组播报文，以通知组内的备份路由器，主路由器处于正常工作状态。(注意：只有Master才发送VRRP报文)，目的IP地址是224.0.0.18代表所有VRRP路由器，报文的TTL是255，协议号是112首先安装抓包工具tcpdumpyum install -y tcpdump正常情况下抓包所谓正常情况即Master和所有Backup都工作正常在所有机器上执行下面抓包命令，-i指定网卡设备为eth0，vrrp指定抓取协议，-n表示显示IP而不是主机名，-vv表示以啰嗦形式显示，即显示更多详细信息tcpdump -i eth0 vrrp -n -vv然后会发现所有机器的VRRP协议包都是一样的，如下04:52:43.041500 IP (tos 0xc0, ttl 255, id 299, offset 0, flags [none], proto VRRP (112), length 40) 172.17.0.2 > 224.0.0.18: vrrp 172.17.0.2 > 224.0.0.18: VRRPv2, Advertisement, vrid 9, prio 100, authtype simple, intvl 1s, length 20, addrs: 172.17.0.100 auth "1111^@^@^@^@"再来对照VRRP协议报文图172.17.0.2：是我们Master的自己的地址224.0.0.18：是VRRP协议规定的组播地址，所谓组播，字面意思，一组机器之间来传播，也就是说，Master发出的数据包所有定义在该组内其他机器都能收到，这也是为什么所有机器抓到的VRRP协议包数据都是一样的，现在只有Master在发VRRP包，其他机器都是在接收而已！ 172.17.0.2 > 224.0.0.18这就能看出来是Master在发包！VRRPv2：表示是VRRP第二个版本的协议Advertisement：Type字段的取值vrid 9：表示Virtual Router ID是9prio 100：Priority是100，和我们配置的优先级一样authtype simple：认证方式为simple，明文认证intvl 1s：VRRP报文通报时间为1s，和我们配置的一致，配置文件中的advert_intaddrs: 172.17.0.100：表示VIP的地址，和我们配置的一致auth "1111^@^@^@^@"：这是我们配置的认证信息从报文我们至少知道现在是172.17.0.2这台机器在接管网关的工作，虚拟IP为172.17.0.100，当然还有优先级之类的信息...Master宕机后抓包首先在备机上抓包，注意，此处抓包略有不同，需要抓网卡上的所有包，不只是vrrp包，这样做的目的是看清楚Master宕机后的arp包，在备机（172.17.0.3）上执行如下抓包命令，很简单，把vrrp限制去掉就可以tcpdump -i eth0 -n -vv然后模拟Master宕机，停止Master上的keepalived，在Master(172.17.0.2)上执行如下命令killall keepalived此时，可以看到备机上的arp包和vrrp包 # Master宕机之前还是正常的 05:04:26.092460 IP (tos 0xc0, ttl 255, id 1002, offset 0, flags [none], proto VRRP (112), length 40) 172.17.0.2 > 224.0.0.18: vrrp 172.17.0.2 > 224.0.0.18: VRRPv2, Advertisement, vrid 9, prio 0, authtype simple, intvl 1s, length 20, addrs: 172.17.0.100 auth "1111^@^@^@^@" # Master宕机会发出了IGMP包 05:04:26.094714 IP (tos 0xc0, ttl 1, id 0, offset 0, flags [DF], proto IGMP (2), length 40, options (RA)) 172.17.0.2 > 224.0.0.22: igmp v3 report, 1 group record(s) [gaddr 224.0.0.18 to_in { }] 05:04:26.608852 IP (tos 0xc0, ttl 1, id 0, offset 0, flags [DF], proto IGMP (2), length 40, options (RA)) 172.17.0.2 > 224.0.0.22: igmp v3 report, 1 group record(s) [gaddr 224.0.0.18 to_in { }] # Master宕机后一系列ARP请求 05:04:26.780921 ARP, Ethernet (len 6), IPv4 (len 4), Request who-has 172.17.0.100 (Broadcast) tell 172.17.0.100, length 28 05:04:26.780965 ARP, Ethernet (len 6), IPv4 (len 4), Request who-has 172.17.0.100 (Broadcast) tell 172.17.0.100, length 28 05:04:26.780968 ARP, Ethernet (len 6), IPv4 (len 4), Request who-has 172.17.0.100 (Broadcast) tell 172.17.0.100, length 28 05:04:26.780971 ARP, Ethernet (len 6), IPv4 (len 4), Request who-has 172.17.0.100 (Broadcast) tell 172.17.0.100, length 28 05:04:26.780974 ARP, Ethernet (len 6), IPv4 (len 4), Request who-has 172.17.0.100 (Broadcast) tell 172.17.0.100, length 28 # Backup接替Master的工作并组播ARRP包 05:04:26.780997 IP (tos 0xc0, ttl 255, id 449, offset 0, flags [none], proto VRRP (112), length 40) 172.17.0.3 > 224.0.0.18: vrrp 172.17.0.3 > 224.0.0.18: VRRPv2, Advertisement, vrid 9, prio 80, authtype simple, intvl 1s, length 20, addrs: 172.17.0.100 auth "1111^@^@^@^@" 05:04:27.781775 IP (tos 0xc0, ttl 255, id 450, offset 0, flags [none], proto VRRP (112), length 40) 172.17.0.3 > 224.0.0.18: vrrp 172.17.0.3 > 224.0.0.18: VRRPv2, Advertisement, vrid 9, prio 80, authtype simple, intvl 1s, length 20, addrs: 172.17.0.100 auth "1111^@^@^@^@" 可以看到，Master宕机时，自己会发出IGMP包，感兴趣的可以自己查查，不再本文讨论范围重点来了，在Master宕机后，备机会收到一些列的ARP请求询问VIP（172.17.0.100）的MAC地址，而且是它自己发出的！# 172.17.0.100发出询问：谁有172.17.0.100 MAC地址请告诉172.17.0.100一下！ 05:04:26.780965 ARP, Ethernet (len 6), IPv4 (len 4), Request who-has 172.17.0.100 (Broadcast) tell 172.17.0.100, length 28它自己还不知道自己的MAC么？对，这个虚拟IP在Master宕机后漂移到Backup上它当然不知道自己的MAC地址，所以这里会先ARP请求获取Backup的MAC地址，然后Backup再不断发出ARRP组播包告知全网我Backup已经接盘了！谁也别想和我抢！参考https://zhuanlan.zhihu.com/p/47434856https://51hcie.com/p7/file/dc/dc_fd_vrrp_0005.htmlhttps://cshihong.github.io/2017/12/18/%E8%99%9A%E6%8B%9F%E8%B7%AF%E7%94%B1%E5%86%97%E4%BD%99%E5%8D%8F%E8%AE%AE-VRRP/

Spring源码分析之IOC容器初始化流程 看现象maven依赖我们只测试IOC容器，因此只需要引入spring-context即可 <dependencies> <!--测试框架--> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>RELEASE</version> <scope>test</scope> </dependency> <!--假数据声场--> <dependency> <groupId>com.github.javafaker</groupId> <artifactId>javafaker</artifactId> <version>1.0.2</version> </dependency> <!--IOC--> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>RELEASE</version> </dependency> </dependencies> 测试用例 /** * 测试spring的ioc容器 */ public class TestSpringIOC { @Test public void test(){ // create and configure beans ApplicationContext context = new ClassPathXmlApplicationContext("ioc/services.xml","ioc/daos.xml"); // retrieve configured instance UserService service = context.getBean("userService", UserService.class); // use configured instance System.out.println(); System.out.println("测试注入-------------------------------------------------"); List<User> userList = service.findUsers(20); userList.stream().forEach(System.out::println); AwareTestService awareTestService = (AwareTestService)context.getBean("awareTestService"); System.out.println(); System.out.println("测试Aware-------------------------------------------------"); awareTestService.test(); AwareTestService awareTestServiceAlias = (AwareTestService)context.getBean("awareTestServiceAlias"); System.out.println(); System.out.println("测试bean别名---------------------------------------------------"); System.out.println(awareTestServiceAlias == awareTestService);//true } } 测试用例里面引入了两个配置文件：services.xml和daos.xml，看看里面写了啥（引用到的类文末有代码清单）services.xml<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd"> <!-- services --> <bean id="userService" class="com.spring.service.impl.UserServiceImpl"> <property name="userDao" ref="userDao"/><!--此处没定义但是在daos.xml里面定义了--> <!-- additional collaborators and configuration for this bean go here --> </bean> <bean id="awareTestService" class="com.spring.service.impl.AwareTestServiceImpl"> </bean> <!--Bean可以定义别名 如果使用注解@Bean，也可以使用别名，详细见@Bean这个注解源码的注释--> <alias name="awareTestService" alias="awareTestServiceAlias"/> <!-- more bean definitions for services go here --> <!--beanProcessor定义--> <bean class="com.spring.config.MyBeanPostProcessor"></bean> </beans>daos.xml<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd"> <!-- daos --> <bean id="userDao" class="com.spring.dao.impl.UserDaoImpl"> </bean> <!-- more bean definitions for daos go here --> </beans>运行结果初始化 before--实例化的bean对象之前:com.spring.dao.impl.UserDaoImpl@3cb1ffe6 userDao 初始化 after...实例化的bean对象之后:com.spring.dao.impl.UserDaoImpl@3cb1ffe6 userDao 初始化 before--实例化的bean对象之前:com.spring.service.impl.UserServiceImpl@43bd930a userService 初始化 after...实例化的bean对象之后:com.spring.service.impl.UserServiceImpl@43bd930a userService 初始化 before--实例化的bean对象之前:com.spring.service.impl.AwareTestServiceImpl@33723e30 awareTestService 初始化 after...实例化的bean对象之后:com.spring.service.impl.AwareTestServiceImpl@33723e30 awareTestService 测试注入------------------------------------------------- User{name='范烨霖', age=17, phone='17647798623', location='瓦房店'} User{name='尹擎苍', age=56, phone='17314101961', location='泸州'} ... 测试Aware------------------------------------------------- Spring中的6大自动注入aware接口 true ... 测试bean别名--------------------------------------------------- true 思考xml文件里面配置了bean，那么Spring是怎么将这些bean初始化到IOC容器的呢？IOC容器具体是什么呢？IOC容器初始化流程开局一张图几大主要的类AbstractBeanDefinitionReader抽象的bean定义reader，不管你是从哪里读取，是从XML，还是注解读取，都提供统一的接口对外暴露读取bean定义的接口，单独抽象了这一层出来屏蔽掉XML或者注解读取AbstractApplicationContext抽象应用上下文，应用了模板方法模式，下面简称应用上下文或上下文AbstractBeanFactory抽象Bean工厂，运用模板方法模式，下面简称bean工厂或工厂ConfigurableEnvironment当前容器运行的环境，里面包含了一些JDK的系统属性等等，可以通过实现EnvironmentAware接口拿到当前的容器环境BeanFactoryPostProcessor里面有一方法postProcessBeanFactory，顾名思义这是在工厂生产bean之前对工厂本身调用的回调函数，千万要和BeanPostProcessor区分开来BeanPostProcessor里面有俩方法postProcessBeforeInitialization和postProcessAfterInitialization，大腿想一想这应该是在bean初始化之前和初始化之后要调用的回调函数ApplicationListener用大腿想一下这玩意儿应该是注册事件用的，在容器初始化的不同阶段调回调函数IOC容器执行流程ApplicationContext context = new ClassPathXmlApplicationContext("services.xml");就这么一句话就初始化了Spring的IOC容器，怎么初始化的？1.初始化顶层对象AbstractApplicationContext这方法public ClassPathXmlApplicationContext(String[] configLocations, boolean refresh, @Nullable ApplicationContext parent)throws BeansException { super(parent); setConfigLocations(configLocations); if (refresh) { refresh(); } }ClassPathXmlApplicationContext一看就是AbstractApplicationContext的子类，这里用了模板方法，其他的各种ApplicationContext公用AbstractApplicationContext中的通用方法，其他要自己实现的子类去实现即可 ，super(parent)这句话不断往父类调，调到AbstractApplicationContext为止，此时加载AbstractApplicationContext并且初始化其中的一些成员变量，比如容器ID，一些Map，一些锁之类的东西题外话：AbstractApplicationContext中的一些锁是这么定义的，取名叫XXXmonitor，实际上就是个空对象，拿给容器启动关闭的时候配合syncronized使用，这也说明了java中的syncronized，锁的就是对象/** Synchronization monitor for the "refresh" and "destroy". */ private final Object startupShutdownMonitor = new Object();2.引入配置文件引入配置就这句话，接收classpath下面的xml文件路径setConfigLocations(configLocations);这句话没啥好说的就赋个值，需要注意的是里面初始化了容器环境，就下面这句话，是在AbstractApplicationContext定义的public ConfigurableEnvironment getEnvironment() { if (this.environment == null) { this.environment = createEnvironment(); } return this.environment; }3.refresh重点来了，refresh是在AbstractApplicationContext中定义的，字面叫刷新，刷新啥？人家源码里面注释都说了：加载或者刷新可能从XML文件，properties文件，数据库等文件的配置并持久化，说白了就是根据配置实例化bean，这里面又有很多步骤，不是简简单单实例化三个字可以表述清楚的源码在此，别人每句话都写上了注释，我们挨个来看synchronized (this.startupShutdownMonitor) { // 3.1 Prepare this context for refreshing. prepareRefresh(); // 3.2 Tell the subclass to refresh the internal bean factory. ConfigurableListableBeanFactory beanFactory = obtainFreshBeanFactory(); // 3.3 Prepare the bean factory for use in this context. prepareBeanFactory(beanFactory); try { // 3.4 Allows post-processing of the bean factory in context subclasses. postProcessBeanFactory(beanFactory); // 3.5 Invoke factory processors registered as beans in the context. invokeBeanFactoryPostProcessors(beanFactory); // 3.6 Register bean processors that intercept bean creation. registerBeanPostProcessors(beanFactory); // 3.7 Initialize message source for this context. initMessageSource(); // 3.8 Initialize event multicaster for this context. initApplicationEventMulticaster(); // 3.9 Initialize other special beans in specific context subclasses. onRefresh(); // 3.10 Check for listener beans and register them. registerListeners(); // 3.11 Instantiate all remaining (non-lazy-init) singletons. finishBeanFactoryInitialization(beanFactory); // 3.12 Last step: publish corresponding event. finishRefresh(); } catch (BeansException ex) { if (logger.isWarnEnabled()) { logger.warn("Exception encountered during context initialization - " + "cancelling refresh attempt: " + ex); } // Destroy already created singletons to avoid dangling resources. destroyBeans(); // Reset 'active' flag. cancelRefresh(ex); // Propagate exception to caller. throw ex; } finally { // Reset common introspection caches in Spring's core, since we // might not ever need metadata for singleton beans anymore... resetCommonCaches(); } }3.1 刷新前的准备工作：prepareRefresh刷新之前的准备工作，准备啥？准备了容器启动时间，容器状态标志位的设置，初始化一些装事件监听器的容器，这些没啥说的，看源码就知道注意：此处有一个回调函数，说的很清楚，准备的时候子类可以干点事情，干子类自己的事情，当然我们这个子类ClassPathXmlApplicationContext比较懒，它啥也没干protected void initPropertySources() { // For subclasses: do nothing by default. }3.2 初始化bean工厂 obtainFreshBeanFactory这个是重点，重量级角色BeanFactory登上历史舞台，看注释：告诉子类让子类刷新内部的BeanFactory，啥意思？就是说这个BeanFactory你子类去实现，我（容器上下文）不管你子类new的啥工厂，反正你给我new好了，用的时候我调用模板方法你子类返回给我，我做一些公共的方法即可，这也是面向接口编程的好处，我不持有具体的引用具体是那个类在new工厂？看下面AbstractRefreshableApplicationContext.refreshBeanFactory()在new这个工厂，这个AbstractRefreshableApplicationContext是ClassPathXmlApplicationContext的父类，看到了吧，它就是一层包一层，一层实现一点点功能，然后组合嵌套起来实现大功能... DefaultListableBeanFactory beanFactory = createBeanFactory(); //new一工厂 beanFactory.setSerializationId(getId());//工厂的序列化id和上下文的id是一样的，getId()就是上下文的id customizeBeanFactory(beanFactory);// loadBeanDefinitions(beanFactory);//重点来了，此处才在通过XML文件读取成bean的定义 synchronized (this.beanFactoryMonitor) { this.beanFactory = beanFactory; } ... AbstractRefreshableApplicationContext#createBeanFactory()方法 protected DefaultListableBeanFactory createBeanFactory() { return new DefaultListableBeanFactory(getInternalParentBeanFactory()); }我们可以看到new了一个DefaultListableBeanFactory工厂，loadBeanDefinitions将XML文件读取成bean定义，此时才真正开始读取XML文件，真正的读取是在XmlBeanDefinitionReader#doLoadBeanDefinitions和doRegisterBeanDefinitions方法中进行的，Spring中，凡是doXXX都是真正干活的方法，怎么读取的具体请看源码，大多都是一些很繁琐的细节，只要知道在这里读取就OK3.3 为当前上下文准备一下bean工厂 prepareBeanFactory配置工厂的标准上下文特征，例如上下文的类加载器和后处理器。这啥意思，就是说这个工厂我现在要准备在上下文用了，你给我初始化一些东西，初始化啥？就是把spring的一些内部对象给放到了bean工厂里面，可见，上下文需要的一些bean也是放在bean工厂里面的注意此处注册了俩BeanPostProcessor，分别是ApplicationContextAwareProcessor和ApplicationListenerDetectorApplicationContextAwareProcessor：在bean初始化之前注入一些实现了特殊的Aware的接口的bean如果你的bean实现了下面六个Aware接口，分别是ApplicationContext、ApplicationEventPublisher、StringValueResolver、Environment、MessageSource、ResourceLoader这些都是spring内部对象，如果使用了他们，将和spring框架耦合，如非特殊情况不建议使用，具体使用方法见AwareTestServiceImpl.java，设置setter即可注入ApplicationListenerDetector：在bean初始化完成之后检测此bean是否是ApplicationListener，如果是的话在上下文里面注册一下忽略一些bean工厂接口，上面那六个Aware接口是bean工厂自己用的，那实例化bean的时候就不管他们了，具体看代码beanFactory.ignoreDependencyInterface(EnvironmentAware.class); beanFactory.ignoreDependencyInterface(EmbeddedValueResolverAware.class); beanFactory.ignoreDependencyInterface(ResourceLoaderAware.class); beanFactory.ignoreDependencyInterface(ApplicationEventPublisherAware.class); beanFactory.ignoreDependencyInterface(MessageSourceAware.class); beanFactory.ignoreDependencyInterface(ApplicationContextAware.class);向bean工厂注册一些依赖，像容器上下文、事件发布器之类的东西注册给bean工厂，怎么注册？就是放到向bean工厂里面的map而已还注册了一些系统需要的单例bean，名字都是写死的，比如environment、systemProperties、systemEnvironment看一下DefaultListableBeanFactory这个bean工厂的源码，里面一堆一堆的map，map用来干啥？用来注册一堆一堆不同类型的bean，具体你看每个map上面的注释，大概就能猜到它是干啥用的总之，这个方法完善了bean工厂，给bean工厂放入了一些自己需要的资源，我们来看下DefaultListableBeanFactory里面都有些啥成员变量//TODO3.4 预处理bean工厂 postProcessBeanFactory预处理什么？谁来预处理？，这就是个空方法，留给具体的上下文实现，注释写的很清楚，这是留给具体上下文即子类来实现的，代码运行到这里说明bean工厂已经初始化好了，但是还没有实例化bean，子类可以在这里注册一些特殊的BeanPostProcessors所以预处理什么？ 抽象上下文的子类想对bean工厂做点什么你可以做，做啥你子类自己应该知道谁来预处理？当然是抽象上下文的子类来做，当然ClassPathXmlApplicationContext比较懒，他啥也没做3.5 调用bean工厂的预处理 invokeBeanFactoryPostProcessors什么叫调用bean工厂的处理？就是说在bean工厂配置好后，bean实例化之前对bean工厂本身要做的一些配置，就在这里进行，你可以看到一堆一堆的Processors，Processors顾名思义就是处理器嘛看下抽象上下文的addBeanFactoryPostProcessor方法，就是往抽象上下文里面不断装BeanFactoryPostProcessor，当然你也可以拿到抽象上下文实例去往里面装自己的BeanFactoryPostProcessor，明白了么。框架的东西就是到处给提供了很多扩展，你可以往里面加自己的逻辑处理这是抽象上下文里面进行处理的，看看具体处理了啥，目标方法为PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(beanFactory, getBeanFactoryPostProcessors());，这就是个静态方法嘛，你可以理解成工具方法// Invoke BeanDefinitionRegistryPostProcessors first, if any. //对BeanDefinitionRegistry做了个特殊处理 if (beanFactory instanceof BeanDefinitionRegistry) { ... } else { // Invoke factory processors registered with the context instance. //对于已经注册了的BeanFactoryPostProcessors，挨个开始应用 invokeBeanFactoryPostProcessors(beanFactoryPostProcessors, beanFactory); } // Do not initialize FactoryBeans here: We need to leave all regular beans // uninitialized to let the bean factory post-processors apply to them! String[] postProcessorNames = beanFactory.getBeanNamesForType(BeanFactoryPostProcessor.class, true, false); // Separate between BeanFactoryPostProcessors that implement PriorityOrdered, // Ordered, and the rest. List<BeanFactoryPostProcessor> priorityOrderedPostProcessors = new ArrayList<>(); List<String> orderedPostProcessorNames = new ArrayList<>(); List<String> nonOrderedPostProcessorNames = new ArrayList<>(); for (String ppName : postProcessorNames) { if (processedBeans.contains(ppName)) { // skip - already processed in first phase above } else if (beanFactory.isTypeMatch(ppName, PriorityOrdered.class)) { priorityOrderedPostProcessors.add(beanFactory.getBean(ppName, BeanFactoryPostProcessor.class)); } else if (beanFactory.isTypeMatch(ppName, Ordered.class)) { orderedPostProcessorNames.add(ppName); } else { nonOrderedPostProcessorNames.add(ppName); } } // First, invoke the BeanFactoryPostProcessors that implement PriorityOrdered. sortPostProcessors(priorityOrderedPostProcessors, beanFactory); invokeBeanFactoryPostProcessors(priorityOrderedPostProcessors, beanFactory); // Next, invoke the BeanFactoryPostProcessors that implement Ordered. List<BeanFactoryPostProcessor> orderedPostProcessors = new ArrayList<>(orderedPostProcessorNames.size()); for (String postProcessorName : orderedPostProcessorNames) { orderedPostProcessors.add(beanFactory.getBean(postProcessorName, BeanFactoryPostProcessor.class)); } sortPostProcessors(orderedPostProcessors, beanFactory); invokeBeanFactoryPostProcessors(orderedPostProcessors, beanFactory); // Finally, invoke all other BeanFactoryPostProcessors. List<BeanFactoryPostProcessor> nonOrderedPostProcessors = new ArrayList<>(nonOrderedPostProcessorNames.size()); for (String postProcessorName : nonOrderedPostProcessorNames) { nonOrderedPostProcessors.add(beanFactory.getBean(postProcessorName, BeanFactoryPostProcessor.class)); } invokeBeanFactoryPostProcessors(nonOrderedPostProcessors, beanFactory); // Clear cached merged bean definitions since the post-processors might have // modified the original metadata, e.g. replacing placeholders in values... beanFactory.clearMetadataCache();仔细看看上面的源码，都有注释的，沉下心去看是个人肯定能看懂的：最开始，初始化一个processedBeans的list，装已经执行过的processors，避免重复执行首先，对BeanDefinitionRegistry这种特殊的工厂特殊处理，这个先略过，咱们这个不是BeanDefinitionRegistry走else逻辑，挨个把从上下文传进来的beanFactoryPostProcessors执行了一遍，这样不就完了吗？还没完，继续看下面beanFactory.getBeanNamesForType从bean工厂里面拿出BeanFactoryPostProcessor！除了上下文传进来的Processor，bean工厂里面可能还有！下面就比较有趣了，创建了仨list，看名字就知道和优先级有关系，就是说执行是有优先级的，优先级高的先执行，优先级低的后执行，没有优先级的最后执行所以就这么个东西。把所有的processors，bean工厂里面的，上下文里面的拿出来执行一遍，虽然代码看起来多，但就干了这么个事3.6 注册BeanPostProcessors registerBeanPostProcessors顾名思义，注册BeanPostProcessors，大胆猜想一些，就是把BeanPostProcessors放到bean工厂的map里面去，注意这里是在注册，而不是执行，看方法前面都有个registerBeanPostProcessors是啥，看它接口的定义，俩方法：postProcessBeforeInitialization、postProcessAfterInitialization，一个是在bean被初始化之前调用，一个是在bean初始化之后调用，说白了就是回调函数，一定要和BeanFactoryPostProcessors区分开来，前者是针对bean的，后者是针对bean工厂的，名字里面都有个Factory嘛，已经很明显了就一句话PostProcessorRegistrationDelegate.registerBeanPostProcessors(beanFactory, this);，里面的逻辑和invokeBeanFactoryPostProcessors差不多，也是把processors分为了带优先级的，带顺序的和不带顺序的，还有一种叫做内部的，把这几种放入bean工厂的map里面，再次强调，此时只是注册，还没有开始执行！！！其实，BeanPostProcessors运用了责任链的设计模式。重写自己的processors参考代码MyBeanPostProcessor，然后在xml文件里面注册这个bean，当然不注册也可以手动在context里面添加，不管怎么样只要能放到bean工厂里面的map就行了多说一句：在xml里面定义的processor会成为bean定义读取到bean工厂中，注册processor的时候有这样一段代码，从bean工厂的bean定义里面拿processor对应的就是xml注册的方式String[] postProcessorNames = beanFactory.getBeanNamesForType(BeanPostProcessor.class, true, false);直接调用addBeanPostProcessor，这就没啥好说的，直接放到bean工厂map里面，上面xml的方式最终其实也是调用了该方法而已3.7 为当前上下文初始化消息源 initMessageSource啥消息源？这是spring为了支持国际化，具体参考https://blog.csdn.net/SLN2432713617/article/details/945928133.8 初始化事件广播 initApplicationEventMulticaster怎么去初始化？有就不管，没有就创建一个嘛，刚开始明显没有，所以创建了一个SimpleApplicationEventMulticaster，SimpleApplicationEventMulticaster是个啥？就是一个事件广播器，啥叫广播器？就是广播一个事件，所有的监听器都去响应这个事件，监听器就是ApplicationListener，注册监听器就是registerListeners做的事情思考一下：为什么要去检查此时有没有事件广播器？去哪里检查？去bean工厂检查为什么要检查？因为我们可以在xml文件中自定义SimpleApplicationEventMulticaster，向下面这样，这样定义的广播器在bean工厂里面，所以要去检查一下<!-- 定义applicationEventMulticaster，注入线程池和errorHandler,此处使用系统自带的广播器，也可以注入其他广播器， --> <bean name="applicationEventMulticaster" class="org.springframework.context.event.SimpleApplicationEventMulticaster"> <property name="taskExecutor" ref="executor"></property> <property name="errorHandler" ref="errorHandler"></property> </bean>其实我们可以在xml中定义很多spring自己的需要的对象，比如上面的广播器，定义的都在bean工厂里面，所以你会看到很多地方都会在bean工厂里面先检查一下有没有已经定义好的事件广播器的最佳实践是用线程池进行异步调用，看下SimpleApplicationEventMulticaster的multicastEvent方法，里面判断了如果有线程池就放到线程池里执行xml里面类似于这样<!-- 定义applicationEventMulticaster，注入线程池和errorHandler,此处使用系统自带的广播器，也可以注入其他广播器， --> <bean name="applicationEventMulticaster" class="org.springframework.context.event.SimpleApplicationEventMulticaster"> <property name="taskExecutor" ref="executor"></property> <property name="errorHandler" ref="errorHandler"></property> </bean>注解类似于这样（springboot）<!-- 定义applicationEventMulticaster，注入线程池和errorHandler,此处使用系统自带的广播器，也可以注入其他广播器， --> <bean name="applicationEventMulticaster" class="org.springframework.context.event.SimpleApplicationEventMulticaster"> <property name="taskExecutor" ref="executor"></property> <property name="errorHandler" ref="errorHandler"></property> </bean>springboor广播器的使用请参考模块springboot-mybatis-plus里面的TestEvent.java3.9 再刷新上下文中调用 onRefresh这又是一个回调方法，前面有一个postProcessBeanFactory也是拿给子类回调的，不细说了3.10 注册事件监听器 registerListeners注册监听器，不是注册给上下文，是注册给广播器，getApplicationEventMulticaster拿到的就是事件广播器，是在initApplicationEventMulticaster中初始化的，说白了就是放在了事件广播器的map里面也就是说，有事件产生的时候，广播器挨个通知所有的监听器，让他们执行相应的动作，怎么通知？一个for循环挨个执行下就好了3.11 重点来了：初始化所有非懒加载的bean finishBeanFactoryInitialization上面铺垫了这么久，现在终于可以开始初始化所有bean了，再次强调，是初始化所有非懒加载！！的bean，用递归的来初始化bean以及bean里面的依赖！核心是反射，里面有一堆一堆的判断条件，还有条件很深的递归，跟代码很容易就晕了，下面演示了空构造方法，里面依赖有一个bean的实例化，即userService，userService里面依赖了userDaoDefaultListableBeanFactory.preInstantiateSingletons：开始挨个实例化bean，是bean工厂在干这事for (String beanName : beanNames) { //这里就在挨个循环处理 ... else { getBean(beanName); } }AbstractBeanFactory.doGetBean：凡是带do的方法都是实际干活儿的方法，doGetBean也不例外，注意这里藏着条件很深的递归，用大腿想想都知道，我要实例化的bean里面依赖另外未实例化的bean，那么我是不是得实例化到最深处？答案是肯定的AbstractBeanFactory.createBean：是定义在AbstractBeanFactory的空方法，子类自己去实现AbstractAutowireCapableBeanFactory.doCreateBean：这才是真正干活的方法，和下面的populateBean是平级的instanceWrapper = createBeanInstance(beanName, mbd, args);AbstractAutowireCapableBeanFactory.createBeanInstance：返回的是BeanWrapper，是实例化bean外面又包装了一层，这里主要是检查了构造方法// No special handling: simply use no-arg constructor. return instantiateBean(beanName, mbd);AbstractAutowireCapableBeanFactory.instantiateBean：用空构造方法实例化beanInstantiationStrategy.instantiate：顾名思义，用这种实例化策略来实例化bean，具体调用的是SimpleInstantiationStrategy.instantiateBeanUtils.instantiateClass：最终调到了这里，这只是一个工具方法，重点就在这里，这里在用反射！！return ctor.newInstance(argsWithDefaultValues);方法定义是这样的，这一看就是用构造方法来实例化bean，这不就是反射么public static T instantiateClass(Constructor ctor, Object... args); 当然这并没有完，这才是构建了它自己，也就是说到这里userService才构建完毕，它里面还有依赖的bean，难道你不初始化？？答案是肯定要AbstractAutowireCapableBeanFactory.populateBean：这里和上面的doCreateBean是平级的AbstractAutowireCapableBeanFactory.applyPropertyValues：将解析到的userDao应用到userServiceBeanDefinitionValueResolver.resolveValueIfNecessary：bean定义的解析器开始解析userDao了BeanDefinitionValueResolver.resolveReference： 重点来了，它又在调bean工厂的getBean方法，里面其实调用了doGetBean！！！开始形成递归了，这一步其实是在初始化userDao！bean = this.beanFactory.getBean(resolvedName);3.12 完成刷新 finishRefresh完成刷新，此处有处理器回调和事件回调protected void finishRefresh() { // Clear context-level resource caches (such as ASM metadata from scanning). clearResourceCaches();// Initialize lifecycle processor for this context. initLifecycleProcessor();// Propagate refresh to lifecycle processor first. getLifecycleProcessor().onRefresh();// Publish the final event. publishEvent(new ContextRefreshedEvent(this));// Participate in LiveBeansView MBean, if active. LiveBeansView.registerApplicationContext(this);}比如，可以做一些在容器刷新完成后的事件：参考```ContextRefreshApplicationListener.java``` ### 3.13 清掉缓存 resetCommonCaches 这个就没啥好说的，不用的东西清理掉而已protected void resetCommonCaches() { ReflectionUtils.clearCache(); AnnotationUtils.clearCache(); ResolvableType.clearCache(); CachedIntrospectionResults.clearClassLoader(getClassLoader());} # Spring IOC BeanPostProcessor是在哪里进行回调的？ 最简单的办法就是写一个`BeanPostProcessor`打断点观察调用堆栈，参考```com.spring.config.MyBeanPostProcessor``` ## postProcessBeforeInitialization的调用堆栈postProcessBeforeInitialization:10, MyBeanPostProcessorapplyBeanPostProcessorsBeforeInitialization:416, AbstractAutowireCapableBeanFactoryinitializeBean:1788, AbstractAutowireCapableBeanFactorydoCreateBean:595, AbstractAutowireCapableBeanFactorycreateBean:517, AbstractAutowireCapableBeanFactorylambda$doGetBean$0:323, AbstractBeanFactorygetObject:-1, 690339675getSingleton:226, DefaultSingletonBeanRegistrydoGetBean:321, AbstractBeanFactorygetBean:202, AbstractBeanFactorypreInstantiateSingletons:895, DefaultListableBeanFactoryfinishBeanFactoryInitialization:878, AbstractApplicationContextrefresh:550, AbstractApplicationContext ## postProcessAfterInitialization调用堆栈postProcessAfterInitialization:17, MyBeanPostProcessorapplyBeanPostProcessorsAfterInitialization:431, AbstractAutowireCapableBeanFactoryinitializeBean:1800, AbstractAutowireCapableBeanFactorydoCreateBean:595, AbstractAutowireCapableBeanFactorycreateBean:517, AbstractAutowireCapableBeanFactorylambda$doGetBean$0:323, AbstractBeanFactorygetObject:-1, 690339675getSingleton:226, DefaultSingletonBeanRegistrydoGetBean:321, AbstractBeanFactorygetBean:202, AbstractBeanFactorypreInstantiateSingletons:895, DefaultListableBeanFactoryfinishBeanFactoryInitialization:878, AbstractApplicationContextrefresh:550, AbstractApplicationContext ## 重点方法 通过观察堆栈，最后的公共调用堆栈为```AbstractAutowireCapableBeanFactory.initializeBean```，拿出来看看长啥样 看到了吧，在调用`invokeInitMethods`之前调用`applyBeanPostProcessorsBeforeInitialization`，之后调用`applyBeanPostProcessorsAfterInitialization`protected Object initializeBean(final String beanName, final Object bean, @Nullable RootBeanDefinition mbd) { ... if (mbd == null || !mbd.isSynthetic()) { wrappedBean = applyBeanPostProcessorsBeforeInitialization(wrappedBean, beanName);} try { invokeInitMethods(beanName, wrappedBean, mbd);} catch (Throwable ex) {...}if (mbd == null || !mbd.isSynthetic()) { wrappedBean = applyBeanPostProcessorsAfterInitialization(wrappedBean, beanName);} ... return wrappedBean;}需要注意的是，如果A依赖B，那么B的`postProcessBeforeInitialization`和`postProcessAfterInitialization`会先调用，然后B创建并返回，此时A的`Before...`和`After...`才会调用，也就是说，从最深层开始调用 # 总结Spring是怎么将这些bean初始化到IOC容器的呢？初始化bean工厂 > 通过BeanDefinitionReader从不同的来源读取配置文件 > 执行针对bean工厂本身的BeanFactoryPostProcessor > 通过反射递归实例化bean(过程中调用BeanPostProcessor) > 建立整个应用上下文 > 一些后续工作(清除缓存，发布事件之类)IOC容器具体是什么呢？所谓IOC容器，其实就是一个Map<String，Object>的数据结构，如下/** Cache of singleton objects: bean name --> bean instance */ private final Map<String, Object> singletonObjects = new ConcurrentHashMap<String, Object>(64);代码清单UserDao与UserDaoImplpublic interface UserDao {List<User> findUsers(int howMany);}public class UserDaoImpl implements UserDao {@Override public List<User> findUsers(int howMany) { List<User> list = new ArrayList<>(); Faker faker = new Faker(new Locale("zh-CN")); for (int i = 0; i < howMany; i++) { User user = new User(); user.setName(faker.name().fullName()); user.setAge(faker.number().numberBetween(0,100)); user.setPhone(faker.phoneNumber().cellPhone()); user.setLocation(faker.address().city()); list.add(user); } return list; }} - `UserService`与`UserServiceImpl` public interface UserService {List<User> findUsers(); List<User> findUsers(int howMany);}public class UserServiceImpl implements UserService {private UserDao userDao; /*如果是配置文件注入要使用setter*/ public void setUserDao(UserDao userDao) { this.userDao = userDao; } @Override public List<User> findUsers() { return userDao.findUsers(10); } @Override public List<User> findUsers(int howMany) { return userDao.findUsers(howMany); } } - `AwareTestService`与`AwareTestServiceImpl` public interface AwareTestService extends EnvironmentAware, EmbeddedValueResolverAware,ResourceLoaderAware,ApplicationEventPublisherAware,MessageSourceAware,ApplicationContextAware {void test();}/**spring框架提供了多个*Aware接口，用于辅助Spring Bean编程访问Spring容器。通过实现这些接口，可以增强Spring Bean的功能，将一些特殊的spring内部bean暴露给业务应用。 * */public class AwareTestServiceImpl implements AwareTestService {ApplicationContext applicationContext; ApplicationEventPublisher applicationEventPublisher; StringValueResolver stringValueResolver; Environment environment; MessageSource messageSource; ResourceLoader resourceLoader; public void test(){ System.out.println("Spring中的6大自动注入aware接口"); System.out.println(this.applicationContext.containsBean("awareTestService")); System.out.println(this.environment.getProperty("java.home")); System.out.println(this.stringValueResolver.resolveStringValue("${java.runtime.version}")); System.out.println(this.resourceLoader.getResource("classpath:src/main/resources/ioc/services.xml")); System.out.println(this.applicationEventPublisher); System.out.println(this.messageSource.getMessage("Message", null, "not exist", Locale.JAPAN)); } @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; } @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; } @Override public void setEmbeddedValueResolver(StringValueResolver resolver) { this.stringValueResolver = resolver; } @Override public void setEnvironment(Environment environment) { this.environment = environment; } @Override public void setMessageSource(MessageSource messageSource) { this.messageSource = messageSource; } @Override public void setResourceLoader(ResourceLoader resourceLoader) { this.resourceLoader = resourceLoader; }} - `MyBeanPostProcessor` public class MyBeanPostProcessor implements BeanPostProcessor {@Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { System.out.println("初始化 before--实例化的bean对象之前:"+bean+"\t"+beanName); return bean; } @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { System.out.println("初始化 after...实例化的bean对象之后:"+bean+"\t"+beanName); return bean; }} - `User` public class User {private String name; private int age; private String phone; private String location; ...}

Linux中对文本分组统计排序 场景有时候需要对一些日志进行分组统计排序，比如统计下nginx日志昨天的QPS是多少原理分析文本，取出目标信息，可以用cut，在进行分组统计，可以用awk，最后排序，可以用sort实战nginx access日志长下面这样192.168.1.100 - - [13/Jan/2021:00:05:11 +0800] "POST /app/images/logo.png HTTP/1.1" 200 3430 "-" "Jakarta Commons-HttpClient/3.1" "-" "0.106" "0.106" "192.168.1.105:8080" 192.168.1.100 - - [13/Jan/2021:00:05:11 +0800] "POST /app/images/logo.png HTTP/1.1" 200 3430 "-" "Jakarta Commons-HttpClient/3.1" "-" "0.106" "0.106" "192.168.1.105:8080"分析一下：要统计QPS，只要把时间13/Jan/2021:00:05:11取出来分组统计就可以了取时间cut用cut命令，-c20-39意思是从第23取到39，不知道具体是多少就要个测，前面最好加一个head -n 10表示取首10行，否则日志太大终端直接卡死head -n 10 access_bak_2021-01-13.log| cut -c20-39 13/Jan/2021:00:05:11 13/Jan/2021:00:05:11 13/Jan/2021:00:05:11 13/Jan/2021:00:05:14 13/Jan/2021:00:05:14 13/Jan/2021:00:05:14 13/Jan/2021:00:05:15 13/Jan/2021:00:05:15 13/Jan/2021:00:05:15 13/Jan/2021:00:05:15分组统计awka[$1]+=1表示对第一列而言，遇到一样的就加一个1，反正对于只有一列的分组统计来说直接加后面这一坨就对了：awk '{a[$1]+=1}END{for (i in a)print i,a[i]}'head -n 10 access_bak_2021-01-13.log| cut -c20-39 | awk '{a[$1]+=1}END{for (i in a)print i,a[i]}' 13/Jan/2021:00:05:11 3 13/Jan/2021:00:05:14 3 13/Jan/2021:00:05:15 4到这里其实统计效果已经出来了排序sortsort命令中-n表示以数值计算，否则sort会认为2比10大，因为2比1大嘛，这是sort的一贯作风-r表示逆序，不加表示顺序-k2表示以第2列来排序当然你也可以写成-nrk2，所以最后就是下面这种效果head -n 10 access_bak_2021-01-13.log| cut -c20-39 | awk '{a[$1]+=1}END{for (i in a)print i,a[i]}' | sort -n -r -k2 13/Jan/2021:00:05:15 4 13/Jan/2021:00:05:14 3 13/Jan/2021:00:05:11 3后记实际中可能是这样的，后面加个| more当数据太多的时候使用cat access_bak_2021-01-13.log| cut -c20-39 | awk '{a[$1]+=1}END{for (i in a)print i,a[i]}' | sort -n -r -k2 | more也可能是这样的，最前面加一个grep用于最基本的筛选，比如你的nginx中多个应用或者你仅仅想统计某个path的时候grep 'app' access_bak_2021-01-13.log| cut -c20-39 | awk '{a[$1]+=1}END{for (i in a)print i,a[i]}' | sort -n -r -k2 | more