Zookeeper系列（二）、核心原理_连接zk服务器是tcp连接吗

上一篇我们介绍了Zookeeper的一些基础知识，本篇来讲解zk内部的一些核心原理，帮助我们更好的理解zk的工作机制。

目录

选举机制

Leader选举流程 

Leader选举原理

Watch机制

会话管理

事务与写流程

Zab协议

---

Zookeeper系列：

Zookeeper系列（一）、基础概念

---

选举机制

Leader选举流程 

Leader选举是保证分布式数据一致性的关键所在。当Zookeeper集群中的一台服务器出现以下两种情况之一时，需要进入Leader选举。

1. 服务器初始化启动。
2. 服务器运行期间无法和Leader保持连接。

服务器启动时期的Leader选举

若进行Leader选举，则至少需要两台机器，这里选取3台机器组成的服务器集群为例。在集群初始化阶段，当有一台服务器Server1启动时，其单独无法进行和完成Leader选举，当第二台服务器Server2启动时，此时两台机器可以相互通信，每台机器都试图找到Leader，于是进入Leader选举过程。选举过程如下：

1. 每个Server发出一个投票。由于是初始情况，Server1和Server2都会将自己作为Leader服务器来进行投票，每次投票会包含所推举的服务器的myid（服务器的唯一标识）和ZXID（事务ID），使用(myid, ZXID)来表示，此时Server1的投票为(1, 0)，Server2的投票为(2, 0)，然后各自将这个投票发给集群中其他机器。
2. 接受来自各个服务器的投票。集群的每个服务器收到投票后，首先判断该投票的有效性，如检查是否是本轮投票、是否来自LOOKING状态的服务器。
3. 处理投票。针对每一个投票，服务器都需要将别人的投票和自己的投票进行PK，PK规则如下：
   1. 优先检查ZXID。ZXID比较大的服务器优先作为Leader，即表示该节点知道的信息更多，数据越新，越能保证数据的恢复。
   2. 如果ZXID相同，那么就比较myid。myid较大的服务器作为Leader服务器。
   3. 对于Server1而言，它的投票是(1, 0)，接收Server2的投票为(2, 0)，首先会比较两者的ZXID，均为0，再比较myid，此时Server2的myid最大，于是更新自己的投票为(2, 0)，然后重新投票，对于Server2而言，其无须更新自己的投票，只是再次向集群中所有机器发出上一次投票信息即可。
4. 统计投票。每次投票后，服务器都会统计投票信息，判断是否已经有机器收到了超过半数的机器的投票，对于Server1、Server2而言，都统计出集群中已经有两台机器接受了(2, 0)的投票信息，此时便认为已经选出了Leader就是server2。
5. 改变服务器状态。一旦确定了Leader，每个服务器就会更新自己的状态，如果是Follower，那么就变更为FOLLOWING，如果是Leader，就变更为LEADING。

过半机制（quorum）：所谓“过半”是指大于集群机器数量的一半，即大于或等于（n/2+1）。但由于Observer不参与投票，因此此处的“集群机器数量”不包含Observer角色节点。

服务器运行时期的Leader选举

在Zookeeper运行期间，Leader与非Leader服务器各司其职，即便当有非Leader服务器宕机或新加入，此时也不会影响Leader，但是一旦Leader服务器挂了，那么整个集群将暂停对外服务，进入新一轮Leader选举，其过程和启动时期的Leader选举过程基本一致。例如还是上面的集群，2台server中已经完成了Leader选举，若此时加入一个新的server3则它会直接变成Follower，如果此时server2作为Leader宕机了，那么就会在server1和server3之间发起新的选举，流程和上面的类似。

Leader选举原理

服务器状态

zk集群中的服务器具有四种状态：

• LOOKING：寻找Leader状态。当服务器处于该状态时，它会认为当前集群中没有Leader，因此需要进入Leader选举状态。
• FOLLOWING：跟随者状态。表明当前服务器角色是Follower。
• LEADING：领导者状态。表明当前服务器角色是Leader。
• OBSERVING：观察者状态。表明当前服务器角色是Observer。

选票的数据结构

上面也提到了每个选票中包含了两个最基本的信息，所推举服务器的myid（服务器的唯一标识）和ZXID（事务ID），选票（Vote）在Zookeeper中包含字段如下

• id：被推举的Leader的myid。
• zxid：被推举的Leader事务ID。
• electionEpoch：逻辑时钟，用来判断多个投票是否在同一轮选举周期中，该值在服务端是一个自增序列，每次进入新一轮的选举后，都会对该值进行加1操作。
• peerEpoch：被推举的Leader的epoch【纪元，类似Raft中的term(任期)】。
• state：当前服务器的状态。

节点间的网络通信

每台zk服务器在启动的过程中，会启动一个QuorumPeerManager，负责各台服务器之间的底层Leader选举过程中的网络通信。

1. 消息队列。QuorumCnxManager内部维护了一系列的队列，用来保存接收到的、待发送的消息以及消息的发送器，除接收队列以外，其他队列都按照myid分组形成队列集合，如一个集群中除了自身还有3台机器，那么就会为这3台机器分别创建一个发送队列，互不干扰。
   1. recvQueue：消息接收队列，用于存放那些从其他服务器接收到的消息。
   2. queueSendMap：消息发送队列，用于保存那些待发送的消息，按照myid进行分组。
   3. senderWorkerMap：发送器集合，每个SenderWorker消息发送器都对应一台远程zk服务器，负责消息的发送，也按照myid进行分组。
   4. lastMessageSent：最近发送过的消息，为每个myid保留最近发送过的一个消息。
2. 建立连接。为了能够相互投票，zk集群中的所有机器都需要两两建立起网络连接。QuorumCnxManager在启动时会创建一个ServerSocket来监听Leader选举的通信端口(默认为3888)。开启监听后，Zookeeper能够不断地接收到来自其他服务器的创建连接请求，在接收到其他服务器的TCP连接请求时，会进行处理。为了避免两台机器之间重复地创建TCP连接，Zookeeper只允许myid大的服务器主动和其他机器建立连接，否则断开连接。在接收到创建连接请求后，服务器通过对比自己和远程服务器的myid值来判断是否接收连接请求。如果当前服务器发现自己的myid更大，那么会断开当前连接，然后自己主动和远程服务器建立连接。一旦连接建立，就会根据远程服务器的myid来创建相应的消息发送器SendWorker和消息接收器RecvWorker，并启动。
3. 消息接收与发送。
   1. 消息接收：由消息接收器RecvWorker负责，由于zk为每个远程服务器都分配一个单独的RecvWorker，因此，每个RecvWorker只需要不断地从这个TCP连接中读取消息，并将其保存到recvQueue队列中。
   2. 消息发送：由于zk为每个远程服务器都分配一个单独的SendWorker，因此，每个SendWorker只需要不断地从对应的消息发送队列中获取出一个消息发送即可，同时将这个消息放入lastMessageSent中。在SendWorker中，一旦Zookeeper发现针对当前服务器的消息发送队列为空，那么此时需要从lastMessageSent中取出一个最近发送过的消息来进行再次发送，这是为了解决接收方在消息接收前或者接收到消息后服务器挂了，导致消息尚未被正确处理。同时，zk能够保证接收方在处理消息时，会对重复消息进行正确的处理。

选举算法核心：FastLeaderElection

选票管理

• sendqueue：选票发送队列，用于保存待发送的选票。
• recvqueue：选票接收队列，用于保存接收到的外部投票。
• WorkerReceiver：选票接收器。其会不断地从QuorumCnxManager中获取其他服务器发来的选举消息，并将其转换成一个选票，然后保存到recvqueue中，在选票接收过程中，如果发现该外部选票的选举轮次小于当前服务器的，那么忽略该外部投票，同时立即发送自己的内部投票。
•  WorkerSender：选票发送器，不断地从sendqueue中获取待发送的选票，并将其传递到底层QuorumCnxManager中。

算法核心

上图展示了FastLeaderElection模块是如何与底层网络I/O进行交互的。Leader选举的基本流程如下：

1. 自增选举轮次。Zookeeper规定所有有效的投票都必须在同一轮次中，在开始新一轮投票时，会首先对逻辑时钟进行自增操作。
2. 初始化选票。在开始进行新一轮投票之前，每个服务器都会初始化自身的选票，并且在初始化阶段，每台服务器都会将自己推举为Leader。
3. 发送初始化选票。完成选票的初始化后，服务器就会发起第一次投票。Zookeeper会将刚刚初始化好的选票放入sendqueue中，由发送器WorkerSender负责发送出去。
4. 接收外部投票。每台服务器会不断地从recvqueue队列中获取外部选票。如果服务器发现无法获取到任何外部投票，那么就会立即确认自己是否和集群中其他服务器保持着有效的连接，如果没有连接，则马上建立连接，如果已经建立了连接，则再次发送自己当前的内部投票。
5. 判断选举轮次。在发送完初始化选票之后，接着开始处理外部投票。在处理外部投票时，会根据选举轮次来进行不同的处理。
   1. 外部投票的选举轮次大于内部投票。若服务器自身的选举轮次落后于该外部投票对应服务器的选举轮次，那么就会立即更新自己的选举轮次(逻辑时钟)，并且清空所有已经收到的投票，然后使用初始化的投票来进行PK以确定是否变更内部投票。最终再将内部投票发送出去。
   2. 外部投票的选举轮次小于内部投票。若服务器接收的外选票的选举轮次落后于自身的选举轮次，那么Zookeeper就会直接忽略该外部投票，不做任何处理，并返回步骤4。
   3. 外部投票的选举轮次等于内部投票。此时可以开始进行选票PK。
6. 选票PK。在进行选票PK时，符合下面任意一个条件就需要变更投票：
   1. 若外部投票中推举的Leader服务器的选举轮次大于内部投票，那么需要变更投票。
   2. 若选举轮次一致，那么就对比两者的ZXID，若外部投票的ZXID大，那么需要变更投票。
   3. 若两者的ZXID一致，那么就对比两者的myid，若外部投票的myid大，那么就需要变更投票。
7. 变更投票。经过PK后，若确定了外部投票优于内部投票，那么就变更投票，即使用外部投票的选票信息来覆盖内部投票，变更完成后，再次将这个变更后的内部投票发送出去。
8. 选票归档。无论是否变更了投票，都会将刚刚收到的那份外部投票放入选票集合recvset中进行归档。recvset用于记录当前服务器在本轮次的Leader选举中收到的所有外部投票（按照服务器的myid区分，如{(1, vote1), (2, vote2)...}）。
9. 统计投票。完成选票归档后，就可以开始统计投票，统计投票是为了统计集群中是否已经有过半的服务器认可了当前的内部投票，如果确定已经有过半服务器认可了该投票，则终止投票。否则返回步骤4。
10. 更新服务器状态。若已经确定可以终止投票，那么就开始更新服务器状态，服务器首选判断当前被过半服务器认可的投票所对应的Leader服务器是否是自己，若是自己，则将自己的服务器状态更新为LEADING，若不是，则根据具体情况来确定自己是FOLLOWING或是OBSERVING。

以上10个步骤就是FastLeaderElection的核心，其中步骤4-9会经过几轮循环，直到有Leader选举产生。

Watch机制

Zookeeper从设计模式角度来理解：是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。

通知类似于数据库中的触发器, 对某个Znode设置Watcher, 当Znode发生变化的时候,WatchManager 会调用对应的Watcher。当Znode发生删除, 修改, 创建, 子节点修改的时候, 对应的Watcher 会得到通知。

Watcher的特点：

• 一次性触发： 一个 Watcher 只会被触发一次, 如果需要继续监听, 则需要再次添加Watcher；

• 事件封装： Watcher得到的事件是被封装过的, 包括三个内容 KeeperState（状态），EventType（事件类型），path（节点路径）；

会话管理

使用客户端来创建一个和zk服务端连接的句柄，这就是一个会话（session）。Session一旦建立，状态就是连接中（CONNECTING）状态，然后客户端会尝试去连接zk服务端，连接成功之后状态变成已连接（CONNECTED）。一般正常情况下只会有这两个状态。不过，还是会发生一些无法恢复的错误/故障，比如：session过期，认证失败，或者客户端关闭连接，这种情况下，session状态会变成关闭（CLOSED）状态。Session会在不同的状态之间进行切换：CONNECTING, CONNECTED, RECONNECTING, RECONNECTED, CLOSE。
sessionID：会话ID，用来唯一标识一个会话，每次客户端创建会话的时候，ZooKeeper都会为其分配一个全局唯一的sessionID；
TimeOut：会话超时时间，如果客户端与服务器之间因为网络闪断导致断开连接，并在TimeOut时间内未连上其他server，则此次会话失效，此次会话创建的临时节点将被清理；
ExpirationTime：下次会话超时时间点。ZooKeeper会为每个会话标记一个下次会话超时时间点，便于对会话进行“分桶管理”，同时也是为了高效低耗的实现会话的超时检查与清理。其值接近于当前时间+TimeOut，但不完全相等，下面会介绍原因。

sessionID的前8位确定所在的机器，后56位使用当前时间的毫秒表示进行随机生成。

分桶管理

上面提到，zk会对每个会话标记一个下次会话超时时间点（ExpirationTime），便于对会话进行“分桶管理”。

每个会话创建完毕后，ZooKeeper就会为其计算ExpirationTime，计算方式大体如下：

ExpirationTime = CurrentTime(当前时间) + SessionTimeOut(会话超时时间)

但图中标识的ExpirationTime并不是以上公式简单的算出来的时间。因为在ZooKeeper的实际实现中，还做了一个处理。ZooKeeper的Leader服务器在运行期间会定时的进行会话超时检查，其时间间隔为ExpirationInterval(默认值2000毫秒)，每隔2000毫秒进行一次会话超时检查。为了方便同时对多个会话进行超时检查，完整的ExpirationTime计算方式如下：

ExpirationTime_ = CurrentTime + SessionTimeOut
ExpirationTime = ( ExpirationTime_/ExpirationInterval + 1 ) * ExpirationInterval

注意：不要使用小学的乘法分配律把小括号给消化掉，其目的是为了保证ExpirationTime是ExpirationInterval的整数倍，带来的好处如下：

1. 提高会话检查的效率。让创建时间临近的会话，分配在一个桶中，实际生产环境中一个服务端会有很多客户端会话，逐个检查过期时间会非常耗时，把它们放在一个桶中批量处理，可以大大提高效率。比如CurrentTime为1547046000、1547046001这样的会话就会被分配在一个桶中。
2. 其次，Leader每隔ExpirationInterval 毫秒进行会话的清理，而刚好 ExpirationTime 这个时间点是会话的失效时间点，如果发现失效，直接清理掉就行了，避免了检查时未失效，但没过几毫秒又失效了这种情况。比如，ExpirationTime 是1547046000，如果在1547045998的时刻检查，发现还有效，但过了2ms之后就无效了。而如果会话超时检查和会话超时时间在同一个时间节点的话，就可以避免这种情况的发生。

心跳机制

1. 客户端会不时地向所连接的ZkServer发送ping消息，ZkServer接收到ping消息，或者任何其它消息的时候，都会将客户端的session_id，session_timeout记录在一个map中。
2. Leader ZkServer会周期性地向所有的follower发送心跳消息，follower接收到ping消息后，会将记录session信息的map作为返回消息，返回给leader，同时清空follower本地的map。 Leader使用这些信息重新计算客户端的超时时间。
3. 一旦在session timout的时间到，leader既没有从其它follower上收集到客户端的session信息，也没有直接接收到该客户端的任何请求，那么该客户端的session就会被关闭。

事务与写流程

概念 

事务：ZooKeeper中，能改变ZooKeeper服务器状态的操作称为事务操作。一般包括数据节点创建与删除、数据内容更新和客户端会话创建与失效等操作。对应每一个事务请求，ZooKeeper 都会为其分配一个全局唯一的事务ID，用 ZXID 表示，通常是一个64位的数字。每一个ZXID对应一次更新操作，从这些ZXID中可以间接地识别出ZooKeeper处理这些事务操作请求的全局顺序。

事务日志：所有事务操作都是需要记录到日志文件中的，可通过 dataLogDir配置文件目录，文件是以写入的第一条事务zxid为后缀，方便后续的定位查找。zk会采取“磁盘空间预分配”的策略，来避免磁盘Seek频率，提升zk服务器对事务请求的影响能力。默认设置下，每次事务日志写入操作都会实时刷入磁盘，也可以设置成非实时（写到内存文件流，定时批量写入磁盘），但那样在断电时会带来丢失数据的风险。事务日志记录的次数达到一定数量后，就会将内存数据库序列化一次，使其持久化保存到磁盘上，序列化后的文件称为"快照文件"。有了事务日志和快照，就可以让任意节点恢复到任意时间点。

数据快照：数据快照是zk数据存储中另一个非常核心的运行机制。数据快照用来记录zk服务器上某一时刻的全量内存数据内容，并将其持久化到指定的磁盘文件中，可通过dataDir配置文件目录。可通过参数snapCount来配置两次快照之间的事务操作个数，zk节点记录完事务日志时，会统计判断是否需要做数据快照（距离上次快照，事务操作次数等于[snapCount/2~snapCount] 中的某个值时，会触发快照生成操作，随机值是为了避免所有节点同时生成快照，导致集群运行缓慢）。

写流程

在 zookeeper 中，客户端会随机连接到 zookeeper 集群中的一个节点，如果是读请求，就直接从当前节点中读取数据，如果是写请求，那么请求会被转发给 leader 提交事务，然后 leader 会广播事务，只要有超过半数节点写入成功，那么写请求就会被提交（类似 2PC 事务）。
所有事务请求必须由一个全局唯一的服务器来协调处理，这个服务器就是 Leader 服务器，其他的服务器就是follower。

步骤如下：

1. 客户端发起一个写请求。
2. 如果是 follower 节点接收到该请求，那么它会将该请求转发给 leader 节点处理。
3. leader 会把这个请求转化成一个事务 Proposal（提议），并把这个 Proposal 分发给集群中的所有 Follower 节点（Observer不会被转发）。
4. Leader 节点需要等待所有 Follower 节点的反馈，一旦超过半数的 Follower 节点进行了正确的反馈（执行事务成功），那么 Leader 就会再次向所有的 Follower 节点发送 commit 消息，要求各个 follower 节点对前面的一个 Proposal 进行提交。
5. leader 节点将最新数据同步给 observer 节点。
6. follower 节点将结果返回给客户端。

Zab协议

Zab协议的全称是 Zookeeper Atomic Broadcast （Zookeeper原子广播）。ZooKeeper 能够保证数据一致性主要依赖于ZAB 协议的消息广播，崩溃恢复和数据同步三个过程。

Zab 协议的特性：

1. Zab 协议需要确保那些已经在 Leader 服务器上提交（Commit）的事务最终被所有的服务器提交。
2. Zab 协议需要确保丢弃那些只在 Leader 上被提出而没有被提交的事务。

消息广播

1. 一个事务请求进来之后，Leader节点会将写请求包装成提议（Proposal）事务，并添加一个全局唯一的 64 位递增事务 ID，Zxid。
2. Leader 节点向集群中其他节点广播Proposal事务，Leader 节点和 Follower 节点是解耦的，通信都会经过一个 FIFO 的消息队列，Leader 会为每一个 Follower 节点分配一个单独的 FIFO 队列，然后把 Proposal 发送到队列中。
3. Follower 节点收到对应的Proposal之后会把它持久到磁盘上，当完全写入之后，发一个ACK给Leader。
4. 当Leader节点收到超过半数Follower节点的ACK之后会提交本地机器上的事务，同时开始广播commit，Follower节点收到 commit 之后，完成各自的事务提交。

消息广播类似一个分布式事务的两阶段提交模式。在这种模式下，无法处理因Leader在发起事务请求后节点宕机带来的数据不一致问题。因此ZAB协议引入了崩溃恢复机制。

崩溃恢复

当整个集群在启动时，或者Leader失联后，ZAB协议就会进入恢复模式，恢复模式的流程如下：

1. 集群通过选举机制产生新的Leader，纪元号加1，开始新纪元；
2. 其他节点从新的Leader同步状态；
3. 过半节点完成状态同步，退出恢复模式，进入消息广播模式；

数据同步流程

Leader选举结束后，进入数据同步流程。数据同步流程以Leader数据为基础，让集群数据达到一致性状态。

1. 新上任的Leader把本地快照加载到内存，并通过日志应用快照之后的所有事务，确保Leader数据库是最新的。
2. Follower和Observer把自身的ZXID和Leader的ZXID进行比较，确定每个节点的同步策略。
3. 根据同步策略，Leader把数据同步到各节点。
4. 每个节点同步结束后，Leader向节点发送NEWLEADER指令。
5. 同步完成的Follower节点返回ACK。
6. 当Leader收到过半节点反馈的ACK时，认为同步完成。
7. Leader向Follower节点发送UPTODATE指令，通知集群同步完成，开始对外服务。

 

希望本文对你有帮助，请点个赞鼓励一下作者吧~ 谢谢！