关于 2PC 提交

当一个事务需要跨多个分布式节点的时候，为了保持事务处理的 ACID 特性，就需要引入一个协调者（TM）来统一调度所有分布式节点的执行逻辑。

案例

有两个参与者和一个协调者，参与者 1 操作成功后，参与者 2 也必须操作成功。参与者 1 和参与者 2 属于两台不同的机器，现在需要跨节点提交事务，也就是分布式事务提交。

参与者成功提交事务

过程：

1. 协调者给每一个参与者发起一个事务提交请求。
2. 各个参与者收到请求后，给出回应：要么执行该事务成功，要么执行该事务失败。
3. 如果所有参与者都回复成功执行该事务，那么协调者发起 commit 请求。
4. 参与者提交事务后，给协调者一个反馈。

某些参与者提交事务失败

过程：

1. 协调者给每一个参与者发起一个事务提交请求。
2. 各个参与者收到请求后，给出回应：要么执行该事务成功，要么执行该事务失败。
3. 参与者 2 执行事务失败，协调者直接给所有参与者发送回滚请求。（只要有一个参与者执行事务失败，那么都要回滚。）
4. 参与者回滚事务后，给协调者一个反馈。

ZooKeeper 的 2PC 提交

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

事务及ZXID

事务是指能够改变 Zookeeper 服务器状态的操作，一般包括数据节点的创建与删除、数据节点内容更新和客户端会话创建与失效等操作。对于每个事务请求，Zookeeper 都会为其分配一个全局唯一的事务 ID，即 ZXID，是一个 64 位的数字，高 32 位表示该事务发生的集群选举周期（集群每发生一次 Leader 选举，值加 1），低 32 位表示该事务在当前选择周期内的递增次序（Leader 每处理一个事务请求，值加 1，发生一次 Leader 选择，低 32 位要清 0）。

参考资料：

1. zookeeper的2PC事务提交

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

1. 服务器初始化启动（集群的每个节点都没有数据 -> 以 SID 的大小为准）
2. 服务器运行期间无法和 Leader 保持连接。（集群的每个节点都有数据，或者 Leader 宕机 -> 以 ZXID 和 SID 的最大值为准）

Leader 选举示例

服务器启动时期的 Leader选举

若进行 Leader 选举，则至少需要 2 台机器，两台的高可用性会差一些，如果 Leader 宕机，就剩下一台，自己没有办法选举。这里选取 3 台机器组成的服务器集群为例。

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

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

服务器运行时期的 Leader 选举

在 ZooKeeper 运行期间，Leader 与非 Leader 服务器各司其职，即便当有非 Leader 服务器宕机或新加入，此时也不会影响 Leader，但是一旦 Leader 服务器挂了，那么整个集群将暂停对外服务，进入新一轮 Leader 选举，其过程和启动时期的 Leader 选举过程基本一致。

假设正在运行的有 Server1、Server2、Server3 三台服务器，当前 Leader 是 Server2，若某一时刻 Leader 挂了，此时便开始 Leader 选举。选举过程如下：

1. 变更状态。 Leader 挂掉后，余下的 Observer 服务器都会将自己的服务器状态变更为 LOOKING，然后开始进入 Leader 选举过程。
2. 每个 Server 会发出一个投票。 在运行期间，每个服务器上的 ZXID 可能不同，此时假定 Server1 的 ZXID 为 123，Server3 的 ZXID 为 122，在第一轮投票中，Server1 和 Server3 都会投自己，产生投票 (1, 123) 和 (3, 122)，然后各自将投票发送给集群中所有机器。
3. 接收来自各个服务器的投票。 与启动时过程相同。
4. 处理投票。 与启动时过程相同，此时，Server1 将会成为 Leader。
5. 统计投票。 与启动时过程相同。
6. 改变服务器的状态。 与启动时过程相同。

FinalizeWait

• Happy Case
  

• Unhappy Case
  

在这种情况之下，node3 不会响应来自 node2 的请求，node2 会在 timeout 之后重试。node2 在 timeout 之前没有办法处理请求。

• Unhappy Case & FinalizeWait
  

一个节点在获得一个 vote 的 quorum 之后，在完成选举之前会等待一段时间。如果在这段时间收到更新的 vote，继续执行选举算法。

Leader 选举算法分析

在 3.4.0 后的 ZooKeeper 的版本只保留了 TCP 版本的 FastLeaderElection 选举算法。当一台机器进入 Leader 选举时，当前集群可能会处于以下两种状态：集群中已经存在 Leader、集群中不存在 Leader。
对于集群中已经存在 Leader 而言，此种情况一般都是某台机器启动得较晚，在其启动之前，集群已经在正常工作，对这种情况，该机器试图去选举 Leader 时，会被告知当前服务器的 Leader 信息，对于该机器而言，仅仅需要和 Leader 机器建立起连接，并进行状态同步即可。
而在集群中不存在 Leader 情况下则会相对复杂，其步骤如下：

1. 第一次投票。 无论哪种导致进行 Leader 选举，集群的所有机器都处于试图选举出一个 Leader 的状态，即 LOOKING 状态，LOOKING 机器会向所有其它机器发送消息，该消息成为投票。投票中包含了 SID（服务器的唯一标识）和 ZXID （事务ID），(SID, ZXID) 形式来标识一次投票信息。
   假定 ZooKeeper 由 5 台机器组成，SID 分别为 1、2、3、4、5，ZXID 分别为 9、9、9、8、8，并且此时 SID 为 2 的机器是 Leader 机器，某一时刻，1、2 所在机器出现故障，因此集群开始进行 Leader 选举。
   在第一次投票时，每台机器都会将自己作为投票对象，于是 SID 为 3、4、5 的机器投票情况分别为 (3,9)、(4,8)、(5,8)。
2. 变更投票。 每台机器发出投票后，也会收到其它机器的投票，每台机器会根据一定规则来处理收到的其它机器的投票，并以此来决定是否需要变更自己的投票，这个规则也是整个 Leader 选举算法的核心所在，其中的术语描述如下：
   • vote_sid 接收到的投票中锁推举的 Leader 服务器的 SID。
   • vote_zxid 接收到的投票中所推举 Leader 服务器的 ZXID。
   • self_sid 当前服务器自己的 SID。
   • self_zxid 当前服务器自己的 ZXID。
     每次对收到的投票的处理，都是对 (vote_sid,vote_zxid) 和 (self_sid, self_zxid) 对比的过程。
     规则一：如果 vote_zxid 大于 self_zxid，就认可当前收到的投票，并再次将该投票发送出去。
     规则二：如果 vote_zxid 小于 self_zxid，那么坚持自己的投票，不做任何变更。
     规则三：如果 vote_zxid 等于 self_zxid，那么久对比两者的 SID，如果 vote_sid 大于 self_sid，那么就认可当前收到的投票，并再次将该投票发送出去。
     规则四：如果 vote_zxid 等于 self_zxid，并且 vote_sid 小于 self_sid，那么坚持自己的投票，不做任何变更。
     综合上面的规则，给出下面的集群变更过程。
     
3. 确定 Leader。 经过第二轮投票后，集群中的每台机器都会再次接收到其它机器的投票，然后开始统计投票，如果一台机器收到了超过半数的相同投票，那么这个投票对应的 SID 的机器即为 Leader。此时 Server3 将成为 Leader。

Leader 选举实现细节

1. 服务器状态
   服务器具有四种状态，分别是 LOOKING、FOLLOWING、LEADING、OBSERVING。

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

2. 投票数据结构
   每个投票中包含了两个最基本的信息，所推举服务器的 SID 和 ZXID，投票（Vote）在 ZooKeeper 中包含字段如下：

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

参考资料

1. Zookeeper选举算法原理

客户端与服务端之间任何交互操作都与会话息息相关，如临时节点的生命周期、客户端请求的顺序执行、Watcher 通知机制等。Zookeeper 的连接与会话就是客户端通过实例化 Zookeeper 对象来实现客户端与服务端创建并保持 TCP 连接的过程。

会话状态

在 Zookeeper 客户端与服务端成功完成连接创建后，就创建了一个会话，Zookeeper 会话在整个运行期间的生命周期中，会在不同的会话状态中之间进行切换，这些状态可以分为 CONNECTING、CONNECTED、RECONNECTING、RECONNECTED、CLOSE 等。

一旦客户端开始创建 Zookeeper 对象，那么客户端状态就会变成 CONNECTING 状态，同时客户端开始尝试连接服务端，连接成功后，客户端状态变为 CONNECTED，通常情况下，由于断网或其他原因，客户端与服务端之间会出现断开情况，一旦碰到这种情况，Zookeeper 客户端会自动进行重连服务，同时客户端状态再次变成 CONNCTING，直到重新连上服务端后，状态又变为 CONNECTED，在通常情况下，客户端的状态总是介于 CONNECTING 和 CONNECTED 之间。但是，如果出现诸如会话超时、权限检查或是客户端主动退出程序等情况，客户端的状态就会直接变更为 CLOSE 状态。

会话创建

Session 是 Zookeeper 中的会话实体，代表了一个客户端会话，其包含了如下四个属性：

• sessionID 会话 ID，唯一标识一个会话，每次客户端创建新的会话时，Zookeeper 都会为其分配一个全局唯一的 sessionID。
• TimeOut 会话超时时间，客户端在构造 Zookeeper 实例时，会配置 sessionTimeout 参数用于指定会话的超时时间，Zookeeper 客户端向服务端发送这个超时时间后，服务端会根据自己的超时时间限制最终确定会话的超时时间。
• TickTime 下次会话超时时间点，为了便于 Zookeeper 对会话实行“分桶策略”管理，同时为了高效低耗地实现会话的超时检查与清理，Zookeeper 会为每个会话标记一个下次会话超时时间点，其值大致等于当前时间加上 TimeOut。
• isClosing 标记一个会话是否已经被关闭，当服务端检测到会话已经超时失效时，会将该会话的 isClosing 标记为”已关闭”，这样就能确保不再处理来自该会话的新情求了。

会话管理

Zookeeper 的会话管理主要是通过 SessionTracker 来负责，其采用了分桶策略（将类似的会话放在同一区块中进行管理）进行管理，以便 Zookeeper 对会话进行不同区块的隔离处理以及同一区块的统一处理。

Zookeeper 将所有的会话都分配在不同的区块中，分配的原则是每个会话的下次超时时间点（ExpirationTime）。ExpirationTime 指该会话最近一次可能超时的时间点。同时，Zookeeper Leader 服务器在运行过程中会定时地进行会话超时检查，时间间隔是 ExpirationInterval，默认为 tickTime 的值，ExpirationTime 的计算时间如下：

1

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

会了保持客户端会话的有效性，客户端会在会话超时时间过期范围内向服务端发送 PING 请求来保持会话的有效性（心跳检测）。同时，服务端需要不断地接收来自客户端的心跳检测，并且需要重新激活对应的客户端会话，这个重新激活过程称为 TouchSession。会话激活不仅能够使服务端检测到对应客户端的存货性，同时也能让客户端自己保持连接状态，其流程如下：

如上图所示，整个流程分为四步：

1. 检查该会话是否已经被关闭。 若已经被关闭，则直接返回即可。
2. 计算该会话新的超时时间 ExpirationTime_New。 使用上面提到的公式计算下一次超时时间点。
3. 获取该会话上次超时时间 ExpirationTime_Old。 计算该值是为了定位其所在的区块。
4. 迁移会话。 将该会话从老的区块中取出，放入 ExpirationTime_New 对应的新区块中。

在上面会话激活过程中，只要客户端发送心跳检测，服务端就会进行一次会话激活，心跳检测由客户端主动发起，以 PING 请求形式向服务端发送，在 Zookeeper 的实际设计中，只要客户端有请求发送到服务端，那么就会触发一次会话激活，以下两种情况都会触发会话激活。

1. 客户端向服务端发送请求，包括读写请求，就会触发会话激活。
2. 客户端发现在 sessionTimeout/3 时间内尚未和服务端进行任何通信，那么就会主动发起 PING 请求，服务端收到该请求后，就会触发会话激活。

对于会话的超时检查而言，Zookeeper 使用 SessionTracker 来负责，SessionTracker 使用单独的线程（超时检查线程）专门进行会话超时检查，即逐个一次地对会话桶中剩下的会话进行清理。如果一个会话被激活，那么 Zookeeper 就会将其从上一个会话桶迁移到下一个会话桶中，如 ExpirationTime 1 的 session n 迁移到 ExpirationTime n 中，此时 ExpirationTime 1 中留下的所有会话都是尚未被激活的，超时检查线程就定时检查这个会话桶中所有剩下的未被迁移的会话，超时检查线程只需要在这些指定时间点（ExpirationTime 1、ExpirationTime 2…）上进行检查即可，这样提高了检查的效率，性能也非常好。

会话清理

当 SessionTracker 的会话超时线程检查出已经过期的会话后，就开始进行会话清理工作，大致可以分为如下七步。

1. 标记会话状态为已关闭。 由于会话清理过程需要一段时间，为了保证在此期间不再处理来自该客户端的请求，SessionTracker 会首先将该会话的 isClosing 标记为 true，这样在会话清理期间接收到该客户端的新情求也无法继续处理了。
2. 发起会话关闭请求。 为了使对该会话的关闭操作在整个服务端集群都生效，Zookeeper 使用了提交会话关闭请求的方式，并立即交付给 PreRequestProcessor 进行处理。
3. 收集需要清理的临时节点。 一旦某个会话失效后，那么和该会话相关的临时节点都需要被清理，因此，在清理之前，首先需要将服务器上所有和该会话相关的临时节点都整理出来。Zookeeper 在内存数据库中会为每个会话都单独保存了一份由该会话维护的所有临时节点集合，在 Zookeeper 处理会话关闭请求之前，若正好有以下两类请求到达了服务端并正在处理中。
   • 节点删除请求，删除的目标节点正好是上述临时节点中的一个。
   • 临时节点创建请求，创建的目标节点正好是上述临时节点中的一个。
     对于第一类请求，需要将所有请求对应的数据节点路径从当前临时节点列表中移出，以避免重复删除，对于第二类请求，需要将所有这些请求对应的数据节点路径添加到当前临时节点列表中，以删除这些即将被创建但是尚未保存到内存数据库中的临时节点。
4. 添加节点删除事务变更。 完成该会话相关的临时节点收集后，Zookeeper 会逐个将这些临时节点转换成”节点删除”请求，并放入事务变更队列 outstandingChanges 中。
5. 删除临时节点。 FinalRequestProcessor 会触发内存数据库，删除该会话对应的所有临时节点。
6. 移除会话。 完成节点删除后，需要将会话从 SessionTracker 中删除。
7. 关闭 NIOServerCnxn。 最后，从 NIOServerCnxnFactory 找到该会话对应的 NIOServerCnxn，将其关闭。

重连

当客户端与服务端之间的网络连接断开时，Zookeeper 客户端会自动进行反复的重连，直到最终成功连接上 Zookeeper 集群中的一台机器。此时，再次连接上服务端的客户端有可能处于以下两种状态之一。

1. CONNECTED 如果在会话超时时间内重新连接上集群中一台服务器 。
2. EXPIRED 如果在会话超时时间以外重新连接上，那么服务端其实已经对该会话进行了会话清理操作，此时会话被视为非法会话。

在客户端与服务端之间维持的是一个长连接，在 sessionTimeout 时间内，服务端会不断地检测该客户端是否还处于正常连接，服务端会将客户端的每次操作视为一次有效的心跳检测来反复地进行会话激活。因此，在正常情况下，客户端会话时一直有效的。然而，当客户端与服务端之间的连接断开后，用户在客户端可能主要看到两类异常：CONNECTION_LOSS（连接断开） 和 SESSION_EXPIRED（会话过期）。

1. CONNECTION_LOSS 此时，客户端会自动从地址列表中重新逐个选取新的地址并尝试进行重新连接，直到最终成功连接上服务器。若客户端在 setData 时出现了 CONNECTION_LOSS 现象，此时客户端会收到 None-Disconnected 通知，同时会抛出异常。应用程序需要捕捉异常并且等待 Zookeeper 客户端自动完成重连，一旦重连成功，那么客户端会收到 None-SyncConnected 通知，之后就可以重试 setData 操作。
2. SESSION_EXPIRED 客户端与服务端断开连接后，重连时间耗时太长，超过了会话超时时间限制后没有成功连上服务器，服务器会进行会话清理，此时，客户端不知道会话已经失效，状态还是 DISCONNECTED，如果客户端重新连上了服务器，此时状态为 SESSION_EXPIRED，用于需要重新实例化 Zookeeper 对象，并且看应用的复杂情况，重新恢复临时数据。
3. SESSION_MOVED 客户端会话从一台服务器转移到另一台服务器，即客户端与服务端 S1 断开连接后，重连上了服务端 S2，此时会话就从 S1 转移到了 S2。当多个客户端使用相同的 sessionId/sessionPasswd 创建会话时，会收到 SessionMovedException 异常。因为一旦有第二个客户端连接上了服务端，就被认为是会话转移了。

参考资料

1. 【分布式】Zookeeper会话

ZooKeeper 提供了分布式数据发布/订阅功能，一个典型的发布/订阅模型系统定义了一种一对多的订阅关系，能让多个订阅者同时监听某一个主题对象，当这个主题对象自身状态变化时，会通知所有订阅者，使它们能够做出相应的处理。
ZooKeeper 中，引入了 Watcher 机制来实现这种分布式的通知功能。ZooKeeper 允许客户端向服务端注册一个 Watcher 监听，当服务端的一些事件触发了这个 Watcher，那么就会向指定客户端发送一个事件通知来实现分布式的通知功能。触发事件种类很多，如：节点创建，节点删除，节点改变，子节点改变等。
总的来说可以概括 Watcher 为以下三个过程：客户端向服务端注册 Watcher、服务端事件发生触发 Watcher、客户端回调 Watcher 得到触发事件情况。

Zookeeper中所有读操作（exists(),getData(),getChildren()）都可以设置 Watch 选项。不同的观察能被触发事件不同:

• 当所观察的 znode 被创建、删除或其数据被更新时，设置在 exists 操作上的观察将被触发。
• 当所观察的 znode 被删除，或其数据被更新时，设置在 getData 操作上的观察将被触发。创建 znode 不会触发 getData 操作上的观察，因为 getData 操作成功执行的前提时 znode 必须已经存在。
• 所观察的 znode 的一个子节点被创建或被删除的时候，或者所观察的 znode 自己被删除的时候，设置在 getChildren 操作上的观察将被触发，可以通过观察事件的类型来判断被删除的是 znode 还是其子节点；NodeDelete 类型代表 znode 被删除，NodeChildrenChanged 类型代表一个子节点被删除。

New ZooKeeper 时注册的 Watcher 叫 default watcher，它不是一次性的，只对 client 的连接状态变化作出反应。

那么要实现 Watch，就必须实现 org.apache.zookeeper.Watcher 接口，并且将实现类的对象传入到可以 Watch 的方法中。
在上述说道的所有读操作中，如果需要 Watcher，我们可以自定义 Watcher，如果是 Boolean 型变量，当为 true 时，则使用系统默认的 Watcher，系统默认的 Watcher 是在 Zookeeper 的构造函数中定义的 Watcher。参数中 Watcher 为空或者 false，表示不启用 Wather。

Watch 机制特点

• 一次性触发： 事件发生触发监听，一个 Watcher event 就会被发送到设置监听的客户端，这种效果是一次性的，后续再次发生同样的事件，不会再次触发。如果还需要关注数据的变化，需要再次注册 Watcher。
• 事件封装： ZooKeeper 使用 WatchedEvent 对象来封装服务端事件并传递。WatchedEvent 包含了每一个事件的三个基本属性：通知状态（keeperState），事件类型（EventType）和节点路径（path）。
• event 异步发送： watcher 的通知事件从服务端发送到客户端是异步的。
• 先注册再触发： Zookeeper 中的 watch 机制，必须客户端先去服务端注册监听，这样事件发送才会触发监听，通知给客户端。

常用操作

注意：

• 在使用 delete 或 setData 操作时必须提供被更新的版本号（可以通过 exists 操作获得）。
• Zookeeper 的写操作是具有原子性的，在写操作没有完成前，Zookeeper 允许客户端读到的数据之后于 Zookeeper 服务的最新状态。
• Zookeeper 中有一个被称为 multi 的操作，用于将多个基本操作集合组成一个操作单元，并确保这些基本操作同时被成功执行，或者同时失败。

事件和状态

同一个事件类型在不同的通知状态中代表的含义有所不同，下表列举了常见的通知状态和事件类型。

虽然说 Zookeeper 需要先注册再触发，但是连接状态事件(type=None, path=null)不需要客户端注册，客户端只要有需要直接处理就行了。

参考资料

1. zookeeper的watcher机制
2. zookeeper（四）：核心原理（Watcher、事件和状态）

Master 选举，就是在众多机器或服务中，选举出一个最终“决定权”的领导者，来独立完成一项任务。比如有一项服务是需要对外提供服务，但是要保证高可用，我们就机会进行服务的多项部署，也就是做了一些备份，提高系统的可用性。一旦我们的主服务挂了，我们可以让其它的备份服务进行重新选举，这样我们就能使整个系统不会因服务的挂掉而造成服务不可用。

思路

在 ZooKeeper中，有两种方式可以实现 Master 选举：

1. 谁先创建 master 临时节点，谁就是 master，当一个 master 挂掉了，master 节点就消失了，别的节点就会监听到，就会继续去创建 master 临时节点，以此类推，利用 Zookeeper 的两个特点（一个节点只能成功创建一次、利用监听的机制）
2. 在 master 下面创建临时有序节点，那个节点最小，那个就是 master，节点挂掉，下面那个临时节点就会监听到上面的临时节点挂掉了，从而取代成为 master，以此类推，（利用 Zookeeper 创建节点临时有序的特性）

实现

Curator

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

/**
 * An example leader selector client.
 * Note that {@link LeaderSelectorListenerAdapter} which has the recommended handling for connection state issues
 * <p>
 * Master 选举参与者
 */
@Slf4j
public class LeaderSelectorParticipant extends LeaderSelectorListenerAdapter implements Closeable {
    private final String name;
    private final LeaderSelector leaderSelector;
    private final AtomicInteger leaderCount = new AtomicInteger();

    public LeaderSelectorParticipant(CuratorFramework client, String path, String name) {
        this.name = name;

        // create a leader selector using the given path for management
        // all participants in a given leader selection must use the same path
        // ExampleClient here is also a LeaderSelectorListener but this isn't required
        leaderSelector = new LeaderSelector(client, path, this);

        // for most cases you will want your instance to requeue when it relinquishes leadership
        // 保证在此实例释放领导权之后还可能获得领导权
        leaderSelector.autoRequeue();
    }

    /**
     * 尝试获取领导权
     */
    public void start() {
        // the selection for this instance doesn't start until the leader selector is started
        // leader selection is done in the background so this call to leaderSelector.start() returns immediately
        leaderSelector.start();
    }

    /**
     * 自动关闭
     */
    @Override
    public void close() {
        // 释放领导权
        leaderSelector.close();
    }

    /**
     * 当你的实例被授予领导权时调用。这个方法在你希望释放领导力之前不应该返回
     */
    @Override
    public void takeLeadership(CuratorFramework client) throws Exception {
        // we are now the leader. This method should not return until we want to relinquish leadership

        final int waitSeconds = 5 * new Random().nextInt(1) + 1;

        log.info("{} is now the leader. Waiting " + waitSeconds + " seconds...", name);
        log.info("{} has been leader {} time(s) before.", name, leaderCount.getAndIncrement());
        try {
            Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
        } catch (InterruptedException e) {
            log.info("{}  was interrupted.", name);
            Thread.currentThread().interrupt();
        } finally {
            log.info("{}  relinquishing leadership.\n", name);
        }
    }
}

/**
 * Master 选举示例
 */
@Slf4j
public class LeaderSelectorExample {
    /**
     * 客户端数量
     */
    private static final int CLIENT_QTY = 10;

    /**
     * 路径
     */
    private static final String PATH = "/examples/leader";

    public static void main(String[] args) throws Exception {
        // all of the useful sample code is in ExampleClient.java

        log.info("Create {} clients, have each negotiate for leadership and then wait a random number of seconds before letting another leader election occur.", CLIENT_QTY);
        log.info("Notice that leader election is fair: all clients will become leader and will do so the same number of times.");

        List<CuratorFramework> clients = Lists.newArrayList();
        List<LeaderSelectorParticipant> examples = Lists.newArrayList();
        TestingServer server = new TestingServer();
        try {
            for (int i = 0; i < CLIENT_QTY; ++i) {
                CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
                clients.add(client);

                LeaderSelectorParticipant example = new LeaderSelectorParticipant(client, PATH, "Client #" + i);
                examples.add(example);

                client.start();
                example.start();
            }

            log.info("Press enter/return to quit\n");
            new BufferedReader(new InputStreamReader(System.in)).readLine();
        } finally {
            log.info("Shutting down...");

            for (LeaderSelectorParticipant leaderSelectorParticipant : examples) {
                CloseableUtils.closeQuietly(leaderSelectorParticipant);
            }
            for (CuratorFramework client : clients) {
                CloseableUtils.closeQuietly(client);
            }

            CloseableUtils.closeQuietly(server);
        }
    }
}

分布式锁是控制分布式系统间同步访问共享资源的一种方式。如果不同的系统或同一个系统的不同主机之间共享了同一个资源，那么访问这些资源的时候，需要使用互斥的手段来防止彼此之间的干扰，以保证一致性，这种情况就需要使用分布式锁。

思路

使用临时顺序 znode 来表示获取锁的请求，创建最小后缀数字 znode 的用户成功拿到锁。

避免羊群效应（herd effect）

把锁请求者按照后缀数字进行排队，后缀数字小的锁请求者先获取锁。如果所有的锁请求者都 watch 锁持有者，当代表锁请求者的 znode 被删除以后，所有的锁请求者都会通知到，但是只有一个锁请求者能拿到锁。这就是羊群效应。

为了避免羊群效应，每个锁请求者 watch 它前面的锁请求者。每次锁被释放，只会有一个锁请求者会被通知到。这样做还让锁的分配更具有公平性，锁的分配遵循先到先得的原则。

实现

bash 命令

1. 终端 1 创建节点：create -e /lock
2. 终端 2 创建节点：create -e /lock，此时会提示：Node already exists: /lock
3. 终端 2 watch 节点：stat -w /lock
4. 终端 1 关闭：quit
5. 终端 2 创建节点：create -e /lock，由于终端 1 会释放掉锁，因此能创建成功

Curator

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

/**
 * Simulates some external resource that can only be access by one process at a time
 */
@Slf4j
public class FakeLimitedResource {
    private final AtomicBoolean inUse = new AtomicBoolean(false);

    public void use() {
        // in a real application this would be accessing/manipulating a shared resource

        if (!inUse.compareAndSet(false, true)) {
            throw new IllegalStateException("Needs to be used by one client at a time");
        }

        try {
            Thread.sleep(3 * new Random().nextInt(1));
        } catch (InterruptedException e) {
            log.error("Sleep with exception", e);
        } finally {
            inUse.set(false);
        }
    }
}

/**
 * 资源调用者
 */
@Slf4j
public class ResourceInvoker {
    /**
     * 公共受限资源
     */
    public FakeLimitedResource resource;
    /**
     * 分布式锁
     */
    private InterProcessMutex lock;
    /**
     * 获取锁之前等待的时间（单位：秒）
     */
    private int waitSeconds;

    public ResourceInvoker(FakeLimitedResource resource, InterProcessMutex lock, int waitSeconds) {
        this.resource = resource;
        this.lock = lock;
        this.waitSeconds = waitSeconds;
    }

    public void invoke() throws Exception {
        String threadName = Thread.currentThread().getName();

        if (waitSeconds <= 0) {
            lock.acquire();
        } else if (!lock.acquire(waitSeconds, TimeUnit.SECONDS)) {
            throw new IllegalStateException("Thread[" + threadName + "] could not acquire the lock");
        }

        try {
            log.info("Thread[{}] had the lock", threadName);
            resource.use();
        } finally {
            log.info("Thread[{}] releasing the lock", threadName);
            // always release the lock in a finally block
            lock.release();
        }
    }
}

/**
 * 分布式锁使用示例
 */
public class LockingExample {
    private final CuratorFramework client;

    public LockingExample(CuratorFramework client) {
        this.client = client;
    }

    public void execute() throws InterruptedException {
        FakeLimitedResource resource = new FakeLimitedResource();
        InterProcessMutex lock = new InterProcessMutex(client, "/examples/lock");
        ResourceInvoker invoker = new ResourceInvoker(resource, lock, 3);

        int poolSize = 5;
        int repetitions = poolSize * 10;
        ExecutorService service = new ThreadPoolExecutor(
                poolSize,
                poolSize,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new CustomThreadFactory("lock"));
        Callable<Void> task = () -> {
            try {
                for (int j = 0; j < repetitions; ++j) {
                    invoker.invoke();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return null;
        };

        for (int i = 0; i < poolSize; ++i) {
            service.submit(task);
        }

        service.shutdown();
        service.awaitTermination(10, TimeUnit.MINUTES);
    }
}

Maven 依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

<dependencies>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
    </dependency>
    <!-- zookeeper API的高层封装，大大简化 zookeeper 客户端编程，添加了例如 zookeeper 连接管理、重试机制等 -->
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
    </dependency>
    <!-- zookeeper 典型应用场景的实现，这些实现是基于 curator-framework -->
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
    </dependency>
</dependencies>

配置类

这里直接使用 curator 的 API：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

@Data
@Component
@ConfigurationProperties(prefix = "zookeeper")
public class ZooKeeperProperties {
    /**
     * ZooKeeper 服务器列表，由英文状态逗号分开的 host:port 字符串组成，每一个都代表一台 ZooKeeper 机器，例如，localhost:2181,localhost:2182,localhost:2183
     */
    private String connectString;

    /**
     * 会话超时时间，单位为毫秒。默认是 60000ms
     */
    private int sessionTimeout;

    /**
     * 连接创建超时时间，单位为毫秒，默认是 15000ms
     */
    private int connectionTimeout;

    /**
     * 已经重试的次数。如果是第一次重试，那么该参数为 0
     */
    private int retryCount;

    /**
     * 从第一次重试开始已经花费的时间，单位为毫秒
     */
    private int elapsedTime;
}

@Slf4j
@Configuration
public class ZooKeeperConfiguration {
    private final ZooKeeperProperties properties;

    public ZooKeeperConfiguration(ZooKeeperProperties properties) {
        this.properties = properties;
    }

    @Bean(initMethod = "start")
    public CuratorFramework curatorFramework() {
        return CuratorFrameworkFactory.newClient(
                properties.getConnectString(),
                properties.getSessionTimeout(),
                properties.getConnectionTimeout(),
                new RetryNTimes(properties.getRetryCount(), properties.getElapsedTime()));
    }
}

使用

1
2

@Autowired
private CuratorFramework client;

1. 创建 network

   1	docker network create --driver bridge --subnet 172.22.0.0/16 --gateway 172.22.0.1 op_net

2. 创建 zookeeper.yml

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49

   version: '3.1' services: zk-1: image: zookeeper restart: always hostname: zk-1 container_name: zk-1 ports: - 2181:2181 environment: ZOO_MY_ID: 1 ZOO_SERVERS: server.1=0.0.0.0:2888:3888 server.2=zk-2:2888:3888 server.3=zk-3:2888:3888 networks: default: ipv4_address: 172.22.0.11 zk-2: image: zookeeper restart: always hostname: zk-2 container_name: zk-2 ports: - 2182:2181 environment: ZOO_MY_ID: 2 ZOO_SERVERS: server.1=zk-1:2888:3888 server.2=0.0.0.0:2888:3888 server.3=zk-3:2888:3888; networks: default: ipv4_address: 172.22.0.12 zk-3: image: zookeeper restart: always hostname: zk-3 container_name: zk-3 ports: - 2183:2181 environment: ZOO_MY_ID: 3 ZOO_SERVERS: server.1=zk-1:2888:3888 server.2=zk-2:2888:3888 server.3=0.0.0.0:2888:3888 networks: default: ipv4_address: 172.22.0.13 networks: default: external: name: op_net

3. 启动 ZooKeeper 集群

   1	docker-compose -f zookeeper.yml up -d

参考资料：

1. docker安装zookeeper集群

利用 ZooKeeper 可以非常方便构建一系列分布式应用中都会涉及到的核心功能。

命名服务

命名服务是指通过指定的名字来获取资源或者服务的地址，利用 ZooKeeper 创建一个全局的路径，即是唯一的路径，这个路径就可以作为一个名字，指向集群中的机器，提供的服务的地址，或者一个远程的对象等等。
ZooKeeper 的命名服务有以下两个应用方面：

• 提供类 JNDI 功能，可以把系统中各种服务的名称、地址以及目录信息存放在 ZooKeeper，需要的时候去 ZooKeeper 中读取。
• 制作分布式的序列号生成器。

数据发布/订阅

数据发布/订阅的一个常见的场景是配置中心：发布者把数据发布到 ZooKeeper 的一个或一系列的节点上，供订阅者进行数据订阅，达到动态获取数据的目的。

配置信息一般有几个特点:

• 数据量小的 KV
• 数据内容在运行时会发生动态变化
• 集群机器共享，配置一致

思路：
ZooKeeper 采用的是推拉结合的方式：

• 推: 服务端会推给注册了监控节点的客户端 Watcher 事件通知
• 拉: 客户端获得通知后，然后主动到服务端拉取最新的数据

数据发布/订阅的另一个常见的场景是服务注册中心，不过由于 ZooKeeper 是基于 CP 原则构建的，因此不适合做服务注册与发现。

分布式锁

分布式锁是控制分布式系统间同步访问共享资源的一种方式。如果不同的系统或同一个系统的不同主机之间共享了同一个资源，那么访问这些资源的时候，需要使用互斥的手段来防止彼此之间的干扰，以保证一致性，这种情况就需要使用分布式锁。

思路：
使用临时顺序 znode 来表示获取锁的请求，创建最小后缀数字 znode 的用户成功拿到锁。

Master 选举

Master 选举，就是在众多机器或服务中，选举出一个最终“决定权”的领导者，来独立完成一项任务。比如有一项服务是需要对外提供服务，但是要保证高可用，我们就机会进行服务的多项部署，也就是做了一些备份，提高系统的可用性。一旦我们的主服务挂了，我们可以让其它的备份服务进行重新选举，这样我们就能使整个系统不会因服务的挂掉而造成服务不可用。

思路：
在 ZooKeeper中，有两种方式可以实现 Master 选举：

1. 谁先创建 master 临时节点，谁就是 master，当一个 master 挂掉了，master 节点就消失了，别的节点就会监听到，就会继续去创建 master 临时节点，以此类推，利用 Zookeeper 的两个特点（一个节点只能成功创建一次、利用监听的机制）
2. 在 master 下面创建临时有序节点，那个节点最小，那个就是 master，节点挂掉，下面那个临时节点就会监听到上面的临时节点挂掉了，从而取代成为 master，以此类推，（利用 Zookeeper 创建节点临时有序的特性）

分布式协调/通知

分布式协调/通知是将不同的分布式组件有机结合起来的关键所在。对于一个在多台机器上部署运行的应用而言，通常需要一个协调者(Coordinator)来控制整个系统的运行流程，例如分布式事务的处理、机器间的相互协调等。同时，引入这样一个协调者，便于将分布式协调的职责从应用中分离出来，从而大大减少系统之间的耦合性，而且能够显著提高系统的可扩展性。

协调/通知机制通常有两种方式：
系统调度模式：操作人员发送通知实际是通过控制台改变某个节点的状态，然后 Zookeeper 将这些变化发送给注册了这个节点的 Watcher 的所有客户端。
工作汇报模式：这个情况是每个工作进程都在某个目录下创建一个临时节点，并携带工作的进度数据。这样汇总的进程可以监控目录子节点的变化获得工作进度的实时的全局情况。

思路：
用 Zookeeper 的 watcher 注册和异步通知功能，通知的发送者创建一个节点，并将通知的数据写入的该节点；通知的接受者则对该节点注册 watch，当节点变化时，就算作通知的到来。

集群管理

所谓集群管理，包括集群监控与集群控制两大块，前者侧重对集群运行时状态的收集，后者则是对集群进行操作与控制。在日常开发和运维过程中，我们经常会有类似于如下的需求：

• 希望知道当前集群中究竟有多少机器在工作
• 对集群中每台机器的运行时状态进行数据收集
• 对集群中机器进行上下线操作

思路：
ZooKeeper 具有以下两大特性：

• 客户端如果对 ZooKeeper 的一个数据节点注册 Watcher 监听，那么当该数据节点的内容或是其子节点列表发生变更时，ZooKeeper 服务器就会向订阅的客户端发送变更通知。
• 对在 ZooKeeper 上创建的临时节点，一旦客户端与服务器之间的会话失效，那么该临时节点也就被自动清除。

利用 ZooKeeper 的这两大特性，我们可以很方便地实现集群机器存活性监控的系统。

负载均衡

负载均衡是一种手段，用来把对某种资源的访问分摊给不同的设备，从而减轻单点的压力。
我们可以自定义一个负载均衡算法，在每个请求过来时从 ZooKeeper 服务器中获取当前集群服务器列表，根据算法选出其中一个服务器来处理请求。

分布式队列

使用 ZooKeeper 来实现分布式队列，分为两大类：

• FIFO 先进先出队列
  队列按照 FIFO 方式进行入队和出队操作，例如实现生产者和消费者模型。

• Barrier 分布式屏障
  当一个队列的成员都聚齐时，这个队列才可用，否则一直等待所有成员到达，这种是同步队列。

思路：

1. 首先利用 Zookeeper 中临时顺序节点的特点
2. 当生产者创建节点生产时，需要判断父节点下临时顺序子节点的个数，如果达到了上限，则阻塞等待；如果没有达到，就创建节点。
3. 当消费者获取节点时，如果父节点中不存在临时顺序子节点，则阻塞等待；如果有子节点，则获取执行自己的业务，执行完毕后删除该节点即可。
4. 获取时获取最小值，保证 FIFO 特性。

虽然 ZooKeeper 可以用来实现分布式队列，但是并不建议使用，原因有如下几点：

1. ZooKeeper 对于传输数据有一个 1MB 的大小限制，这就意味着实际中 ZooKeeper 节点 ZNodes 必须设计的很小，但实际中队列通常都存放着数以千计的消息。
2. 如果有很多大的 ZNodes 会严重拖慢的 ZooKeeper 启动过程，包括 ZooKeeper 节点之间的同步过程，如果真要用 ZooKeeper 当队列，最好去调整 initLimit 与 syncLimit 参数。
3. 如果一个 ZNode 过大，也会导致清理变得困难，也会导致 getChildren() 方法失败，Netflix 不得不设计一个特殊的机制来处理这个大体积的 ZNode。
4. 如果 ZooKeeper 中某个 node 下有数千子节点，也会严重拖累 ZooKeeper 性能。
5. ZooKeeper 中的数据都会放置在内存中。

参考资料

1. ZooKeeper 的应用场景

• 每个 Server 在内存中存储了一份数据
• ZooKeeper 启动时，将从实例中选举一个 leader（Paxos 协议）
• Leader 负责处理数据更新等操作（Zab 协议）
• 一个更新操作成功，当且仅当大多数 Server 在内存中成功修改数据

ZooKeeper 本身可以是单机模式，也可以是集群模式，为了 ZooKeeper 本身不出现单点故障，通常情况使用集群模式（Server 数目一般为奇数），而且是 master/slave 模式的集群。

角色

Leader

Leader 不直接接受 Client 的请求，但接受由其他 Follower 和 Observer 转发过来的 Client 请求，此外，Leader 还负责投票的发起和决议，即时更新状态和数据。

Follower

Follower 角色接受客户端请求并返回结果，参与 Leader 发起的投票和选举，但不具有写操作的权限。

Observer

Observer 角色接受客户端连接，将写操作转给 Leader，但 Observer 不参与投票（即不参加一致性协议的达成），只同步 Leader 节点的状态，Observer 角色是为集群系统扩展而生的。

应用场景：

• 提高集群的读性能（未参与事务的提交过程）
• 跨数据中心部署

Follower vs Observer 请求流程

Follower-Leader 写请求流程：

1. 节点 1（Follower） 收到写请求，转发到节点 2（Leader）
2. 节点 2 发送 Propose 给集群中所有 Follower 节点
3. Follower 节点收到 Propose 后，返回 Accept 给 Leader
4. Leader 收到大多数节点的 Accept 后，向所有节点发送 Commit
5. 节点 1 收到 Commit 后，返回客户端，告诉客户端写成功

Observer-Leader 写请求流程：

1. 节点 1（Observer） 收到写请求，转发到节点 2（Leader）
2. 节点 2 发送 Propose 给集群中所有 Follower 节点
3. Follower 节点收到 Propose 后，返回 Accept 给 Leader
4. Leader 收到大多数节点的 Accept 后，向所有节点发送 Commit
5. 节点 1 不参与事务的提交过程，而是一直等待，等到收到 Leader 的 INFORM 后，返回客户端，告诉客户端写成功

节点数

ZooKeeper 要求集群节点数大于 1，且为单数，原因有以下两点。

容错

由于在增删改操作中需要半数以上服务器通过，来分析以下情况：

• 2 台服务器，至少 2 台正常运行才行（2 的半数为 1，半数以上最少为 2），正常运行 1 台服务器都不允许挂掉
• 3 台服务器，至少 2 台正常运行才行（3 的半数为 1.5，半数以上最少为 2），正常运行可以允许 1 台服务器挂掉
• 4 台服务器，至少 3 台正常运行才行（4 的半数为 2，半数以上最少为 3），正常运行可以允许 1 台服务器挂掉
• 5 台服务器，至少 3 台正常运行才行（5 的半数为 2.5，半数以上最少为 3），正常运行可以允许 2 台服务器挂掉
• 6 台服务器，至少 3 台正常运行才行（6 的半数为 3，半数以上最少为 4），正常运行可以允许 2 台服务器挂掉

通过以上可以发现，3 台服务器和 4 台服务器都最多允许 1 台服务器挂掉，5 台服务器和 6 台服务器都最多允许 2 台服务器挂掉，但是明显 4 台服务器成本高于 3 台服务器成本，6 台服务器成本高于 5 服务器成本。这是由于半数以上投票通过决定的。

防脑裂

一个 ZooKeeper 集群中，可以有多个 follower、observer 服务器，但是必需只能有一个 leader 服务器。如果 leader 服务器挂掉了，剩下的服务器集群会通过半数以上投票选出一个新的 leader 服务器。

集群互不通讯情况：

• 一个集群 3 台服务器，全部运行正常，但是其中 1 台裂开了，和另外 2 台无法通讯。3 台机器里面 2 台正常运行过半票可以选出一个 leader。
• 一个集群 4 台服务器，全部运行正常，但是其中 2 台裂开了，和另外 2 台无法通讯。4 台机器里面 2 台正常工作没有过半票以上达到3，无法选出 leader 正常运行。
• 一个集群 5 台服务器，全部运行正常，但是其中 2 台裂开了，和另外 3 台无法通讯。5 台机器里面 3 台正常运行过半票可以选出一个 leader。
• 一个集群 6 台服务器，全部运行正常，但是其中 3 台裂开了，和另外 3 台无法通讯。6 台机器里面 3 台正常工作没有过半票以上达到 4，无法选出 leader 正常运行。

通可以上分析可以看出，为什么 ZooKeeper 集群数量总是单出现，主要原因还是在于第 2 点，防脑裂，对于第 1 点，无非是正常控制，但是不影响集群正常运行。但是出现第 2 种裂的情况，ZooKeeper 集群就无法正常运行了。