zookeeper 使用场景

1、集群配置数据管理
2、配置中心(soa)
3、简单消息推送
4、分布式锁

zookeeper zab协议

ZAB：ZooKeeper的Atomic Broadcast协议，能够保证发给各副本的消息顺序相同。
Zookeeper使用了一种称为Zab（ZookeeperAtomic Broadcast）的协议作为其一致性复制的核心，其特点为高吞吐量、低延迟、健壮、简单，但不过分要求其扩展性。

Zookeeper的实现是有Client、Server构成，Server端提供了一个一致性复制、存储服务，Client端会提供一些具体的语义，比如分布式锁、选举算法、分布式互斥等。从存储内容来说，Server端更多的是存储一些数据的状态，而非数据内容本身，因此Zookeeper可以作为一个小文件系统使用。数据状态的存储量相对不大，完全可以全部加载到内存中，从而极大地消除了通信延迟。
Server可以Crash后重启，考虑到容错性，Server必须“记住”之前的数据状态，因此数据需要持久化，但吞吐量很高时，磁盘的IO便成为系统瓶颈，其解决办法是使用缓存，把随机写变为连续写。

考虑到Zookeeper主要操作数据的状态，为了保证状态的一致性，Zookeeper提出了两个安全属性（Safety Property）：
全序（Total order）：如果消息a在消息b之前发送，则所有Server应该看到相同的结果
因果顺序（Causal order）：如果消息a在消息b之前发生（a导致了b），并被一起发送，则a始终在b之前被执行。
为了保证上述两个安全属性，Zookeeper使用了TCP协议和Leader。通过使用TCP协议保证了消息的全序特性（先发先到），通过Leader解决了因果顺序问题：先到Leader的先执行。因为有了Leader，Zookeeper的架构就变为：Master-Slave模式，但在该模式中Master（Leader）会Crash，因此，Zookeeper引入了Leader选举算法，以保证系统的健壮性。归纳起来Zookeeper整个工作分两个阶段：
Atomic Broadcast
Leader选举
1.Atomic Broadcast
同一时刻存在一个Leader节点，其他节点称为“Follower”，如果是更新请求，如果客户端连接到Leader节点，则由Leader节点执行其请求；如果连接到Follower节点，则需转发请求到Leader节点执行。但对读请求，Client可以直接从Follower上读取数据，如果需要读到最新数据，则需要从Leader节点进行，Zookeeper设计的读写比例是2：1。

Leader通过一个简化版的二段提交模式向其他Follower发送请求，但与二段提交有两个明显的不同之处：
• 因为只有一个Leader，Leader提交到Follower的请求一定会被接受（没有其他Leader干扰）
• 不需要所有的Follower都响应成功，只要一个多数派即可
通俗地说，如果有2f+1个节点，允许f个节点失败。因为任何两个多数派必有一个交集，当Leader切换时，通过这些交集节点可以获得当前系统的最新状态。如果没有一个多数派存在（存活节点数小于f+1）则，算法过程结束。但有一个特例：
如果有A、B、C三个节点，A是Leader，如果B Crash，则A、C能正常工作，因为A是Leader，A、C还构成多数派；如果A Crash则无法继续工作，因为Leader选举的多数派无法构成。

2.Leader Election
Leader选举主要是依赖Paxos算法，Leader选举遇到的最大问题是，”新老交互“的问题，新Leader是否要继续老Leader的状态。这里要按老Leader Crash的时机点分几种情况：
1.老Leader在COMMIT前Crash（已经提交到本地）
2.老Leader在COMMIT后Crash，但有部分Follower接收到了Commit请求
第一种情况，这些数据只有老Leader自己知道，当老Leader重启后，需要与新Leader同步并把这些数据从本地删除，以维持状态一致。
第二种情况，新Leader应该能通过一个多数派获得老Leader提交的最新数据，老Leader重启后，可能还会认为自己是Leader，可能会继续发送未完成的请求，从而因为两个Leader同时存在导致算法过程失败，解决办法是把Leader信息加入每条消息的id中，Zookeeper中称为zxid，zxid为一64位数字，高32位为leader信息又称为epoch，每次leader转换时递增；低32位为消息编号，Leader转换时应该从0重新开始编号。通过zxid，Follower能很容易发现请求是否来自老Leader，从而拒绝老Leader的请求。
因为在老Leader中存在着数据删除（情况1），因此Zookeeper的数据存储要支持补偿操作，这也就需要像数据库一样记录log。

3.Zab与Paxos
Zab的作者认为Zab与paxos并不相同，Zab就是Paxos的一种简化形式，Paxos保证不了全序顺序。
这里首要一点是Paxos的一致性不能达到ZooKeeper的要求。举个例子：假设一开始Paxos系统中的leader是P1，他发起了两个事务<t1, v1>（表示序号为t1的事务要写的值是v1）和<t2, v2>，过程中挂了。新来个leader是P2，他发起了事务<t1, v1’>。而后又来个新leader是P3，他汇总了一下，得出最终的执行序列<t1, v1’>和<t2, v2>，即P2的t1在前，P1的t2在后。
分析为什么不满足ZooKeeper需求：
ZooKeeper是一个树形结构，很多操作都要先检查才能确定能不能执行，比如P1的事务t1可能是创建节点“/a”，t2可能是创建节点“/a/aa”，只有先创建了父节点“/a”，才能创建子节点“/a/aa”。而P2所发起的事务t1可能变成了创建“/b”。这样P3汇总后的序列是先创建“/b”再创建“/a/aa”，由于“/a”还没建，创建“a/aa”就搞不定了。
解决方案：
为了保证这一点，ZAB要保证同一个leader的发起的事务要按顺序被apply，同时还要保证只有先前的leader的所有事务都被apply之后，新选的leader才能在发起事务。

ZAB的核心思想：形象的说就是保证任意时刻只有一个节点是leader，所有更新事务由leader发起去更新所有复本（称为follower），更新时用的就是两阶段提交协议，只要多数节点prepare成功，就通知他们commit。各follower要按当初leader让他们prepare的顺序来apply事务。因为ZAB处理的事务永远不会回滚，ZAB的2PC做了点优化，多个事务只要通知zxid最大的那个commit，之前的各follower会统统commit。
这里有几个关键点：
1、leader和follower之间通过心跳来检测异常
2、检测到异常之后的节点若试图成为新的leader，首先要获得大多数节点的支持，然后从状态最新的节点同步事务，完成后才可正式成为leader发起事务
3、区分新老leader的关键是一个会一直增长的epoch

zookeeper 选举流程

阶段(选举 –> 恢复 –> 广播)

ZAB协议定义了选举（election）、发现（discovery）、同步（sync）、广播(Broadcast) 四个阶段；ZAB选举（election）时当Follower存在ZXID（事务ID）时判断所有Follower节点的事务日志，只有lastZXID的节点才有资格成为Leader，这种情况下选举出来的Leader总有最新的事务日志，基于这个原因所以ZooKeeper实现的时候把 发现（discovery）与同步（sync）合并为恢复（recovery） 阶段；

1.Election：在Looking状态中选举出Leader节点，Leader的lastZXID总是最新的;
2.Discovery：Follower节点向准Leader推送FOllOWERINFO，该信息中包含了上一周期的epoch，接受准Leader的NEWLEADER指令，检查newEpoch有效性，准Leader要确保Follower的epoch与ZXID小于或等于自身的;
3.sync：将Follower与Leader的数据进行同步，由Leader发起同步指令，最总保持集群数据的一致性;
5.Recovery：在Election阶段选举出Leader后本阶段主要工作就是进行数据的同步，使Leader具有highestZXID，集群保持数据的一致性;
4.Broadcast：Leader广播Proposal与Commit，Follower接受Proposal与Commit;

zookeeper ACL

一个ZooKeeper 的节点存储两部分内容：数据和状态，状态中包含ACL 信息。创建一个znode 会产生一个ACL 列表，列表中每个ACL 包括：

① 权限perms

② 验证模式scheme

③ 具体内容expression：Ids

例如，当scheme=”digest” 时， Ids 为用户名密码， 即”root ：J0sTy9BCUKubtK1y8pkbL7qoxSw”。ZooKeeper 提供了如下几种验证模式：

① Digest： Client 端由用户名和密码验证，譬如user:pwd

② Host： Client 端由主机名验证，譬如localhost

③ Ip：Client 端由IP 地址验证，譬如172.2.0.0/24

④ World ：固定用户为anyone，为所有Client 端开放权限

当会话建立的时候，客户端将会进行自我验证。

权限许可集合如下：

① Create 允许对子节点Create 操作

② Read 允许对本节点GetChildren 和GetData 操作

③ Write 允许对本节点SetData 操作

④ Delete 允许对子节点Delete 操作

⑤ Admin 允许对本节点setAcl 操作

另外，ZooKeeper Java API支持三种标准的用户权限，它们分别为：

① ZOO_PEN_ACL_UNSAFE：对于所有的ACL来说都是完全开放的，任何应用程序可以在节点上执行任何操作，比如创建、列出并删除子节点。
② ZOO_READ_ACL_UNSAFE：对于任意的应用程序来说，仅仅具有读权限。
③ ZOO_CREATOR_ALL_ACL：授予节点创建者所有权限。需要注意的是，设置此权限之前，创建者必须已经通了服务器的认证。

Znode ACL 权限用一个int 型数字perms 表示，perms 的5 个二进制位分别表示setacl、delete、create、write、read。比如adcwr=0x1f，—-r=0x1，a-c-r=0x15。

注意的是，exists操作和getAcl操作并不受ACL许可控制，因此任何客户端可以查询节点的状态和节点的ACL。

ACL代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import org.apache.Zookeeper.*;
import org.apache.Zookeeper.server.auth.DigestAuthenticationProvider;
import org.apache.Zookeeper.data.*;
import java.util.*;

public class NewDigest {
    public static void main(String[] args) throws Exception {
        //new一个acl
        List<ACL> acls = new ArrayList<ACL>();
　　　　　//添加第一个id，采用用户名密码形式
        Id id1 = new Id("digest",DigestAuthenticationProvider.generateDigest("admin:admin"));
        ACL acl1 = new ACL(ZooDefs.Perms.ALL, id1);
        acls.add(acl1);
　　　　 //添加第二个id，所有用户可读权限
        Id id2 = new Id("world", "anyone");
        ACL acl2 = new ACL(ZooDefs.Perms.READ, id2);
        acls.add(acl2);
        // Zk用admin认证，创建/test ZNode。
　　　　　ZooKeeper Zk = new ZooKeeper("host1:2181,host2:2181,host3:2181",2000, null);
　　　　　Zk.addAuthInfo("digest", "admin:admin".getBytes());
　　　　　Zk.create("/test", "data".getBytes(), acls, CreateMode.PERSISTENT);
   }
}

ACL超级权限

superDigest 设置的步骤：

① 启动ZK 的时候( zkServer.sh ) ， 加入参数： Java”-Dzookeeper .DigestAuthenticationProvider.superDigest=super:D/InIHSb7yEEbrWz8b9l71RjZJU=” （无空格）。

② 在客户端使用的时候， addAuthInfo(“digest”, “super:test”, 10, 0, 0); “ super:test” 为”super:D/InIHSb7yEEbrWz8b9l71RjZJU=”的明文表示，加密算法同setAcl。

zookeeper 客户端

目前Curator有2.x.x和3.x.x两个系列的版本，支持不同版本的Zookeeper。其中Curator 2.x.x兼容Zookeeper的3.4.x和3.5.x。Curator 3.x.x只兼容Zookeeper 3.5.x，并且提供了一些诸如动态重新配置、watch删除等新特性。

zookeeper 文件与快照

日志路径:/datalog/version-2
快照路径:/data/version-2

1、当开始第一次运行的时候，首先会产生一个snapshot，这个时候是没有log的。当第一次有操作的时候开始log文件产生的时候，这时候不会产生snapshot。
当第一个log文件体积达到预设值时候，这时候系统需要snapshot，然后创建新的日志文件，以此不断交替。
2、生成snapshot触发条件: logcount > (snapcount/2 + randroll) randroll是随机数范围在[1, snapcount/2]

zookeeper网络相关

1、zookeeper的网络请求与连接实现方式有两种，一种是通过netty实现，另外一种是通过Java原生NIO实现。

zookeeper会话管理

session连接

每个ZooKeeper客户端的配置中都包括集合体中服务器的列表。在启动时，客户端会尝试连接到列表中的一台服务器。如果连接失败，它会尝试连接另一台服务器，以此类推，直到成功与一台服务器建立连接或因为所有ZooKeeper服务器都不可用而失败。

一旦客户端与一台ZooKeeper服务器建立连接，这台服务器就会为该客户端创建一个新的会话。每个会话都会有一个超时的时间设置，这个设置由创建会话的应用来设定。如果服务器在超时时间段内没有收到任何请求，则相应的会话会过期。一旦一个会话已经过期，就无法重新打开，并且任何与该会话相关联的短暂znode都会丢失。会话通常长期存在，而且会话过期是一种比较罕见的事件，但对应用来说，如何处理会话过期仍是非常重要的。

只要一个会话空闲超过一定时间，都可以通过客户端发送ping请求（也称为心跳）保持会话不过期。ping请求由ZooKeeper的客户端库自动发送，因此在我们的代码中不需要考虑如何维护会话。这个时间长度的设置应当足够低，以便能档检测出服务器故障（由读超时体现），并且能够在会话超时的时间段内重新莲接到另外一台服务器。

故障切换

ZooKeeper客户端可以自动地进行故障切换，切换至另一台ZooKeeper服务器。并且关键的一点是，在另一台服务器接替故障服务器之后，所有的会话和相关的短暂Znode仍然是有效的。在故障切换过程中，应用程序将收到断开连接和连接至服务的通知。当客户端断开连接时，观察通知将无法发送；但是当客户端成功恢复连接后，这些延迟的通知会被发送。当然，在客户端重新连接至另一台服务器的过程中，如果应用程序试图执行一个操作，这个操作将会失败。这充分体现了在真实的ZooKeeper应用中处理连接丢失异常的重要性。

zookeeper运维命令

1、zookeeper的命令分为两部分，一块是客户端命令，一块是服务端命令,即四字命令,指的是服务端的命令。

2、客户端命令

3、服务端命令

4、用法举例

1
2
3
4
5
6
7
8
9
10
11

echo stat | nc 127.0.0.1 2181 查看哪个节点被选择作为follower或者leader
echo ruok | nc 127.0.0.1 2181 测试是否启动了该Server，若回复imok表示已经启动。
echo dump | nc 127.0.0.1 2181 列出未经处理的会话和临时节点。
echo kill | nc 127.0.0.1 2181 关掉server
echo conf | nc 127.0.0.1 2181 输出相关服务配置的详细信息。
echo cons | nc 127.0.0.1 2181 列出所有连接到服务器的客户端的完全的连接 / 会话的详细信息。
echo envi | nc 127.0.0.1 2181 输出关于服务环境的详细信息（区别于 conf 命令）。
echo reqs | nc 127.0.0.1 2181 列出未经处理的请求。
echo wchs | nc 127.0.0.1 2181 列出服务器watch的详细信息。
echo wchc | nc 127.0.0.1 2181 通过session列出服务器watch的详细信息，它的输出是一个与 watch 相关的会话的列表。
echo wchp | nc 127.0.0.1 2181 通过路径列出服务器watch的详细信息。它输出一个与 session 相关的路径。

5、启停命令

1
2
3

./zkServer.sh restart
./zkServer.sh start
./zkServer.sh stop

6、启用全部四字命令

<1>在conf目录下创建文件java.env(如果不存在的话)

<2>添加语句: export JVMFLAGS=”-Dzookeeper.4lw.commands.whitelist=* $JVMFLAGS”

zookeeper 实现分布式锁

大致思想即为：每个客户端对某个方法加锁时，在zookeeper上的与该方法对应的指定节点的目录下，生成一个唯一的瞬时有序节点。 判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。 当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。

锁无法释放？使用Zookeeper可以有效的解决锁无法释放的问题，因为在创建锁的时候，客户端会在ZK中创建一个临时节点，一旦客户端获取到锁之后突然挂掉（Session连接断开），那么这个临时节点就会自动删除掉。其他客户端就可以再次获得锁。
非阻塞锁？使用Zookeeper可以实现阻塞的锁，客户端可以通过在ZK中创建顺序节点，并且在节点上绑定监听器，一旦节点有变化，Zookeeper会通知客户端，客户端可以检查自己创建的节点是不是当前所有节点中序号最小的，如果是，那么自己就获取到锁，便可以执行业务逻辑了。
不可重入？使用Zookeeper也可以有效的解决不可重入的问题，客户端在创建节点的时候，把当前客户端的主机信息和线程信息直接写入到节点中，下次想要获取锁的时候和当前最小的节点中的数据比对一下就可以了。如果和自己的信息一样，那么自己直接获取到锁，如果不一样就再创建一个临时的顺序节点，参与排队。
单点问题？使用Zookeeper可以有效的解决单点问题，ZK是集群部署的，只要集群中有半数以上的机器存活，就可以对外提供服务。

使用ZK实现的分布式锁好像完全符合了本文开头我们对一个分布式锁的所有期望。但是，其实并不是，Zookeeper实现的分布式锁其实存在一个缺点，那就是性能上可能并没有缓存服务那么高。因为每次在创建锁和释放锁的过程中，都要动态创建、销毁瞬时节点来实现锁功能。ZK中创建和删除节点只能通过Leader服务器来执行，然后将数据同步到所有的Follower机器上。

其实，使用Zookeeper也有可能带来并发问题，只是并不常见而已。考虑这样的情况，由于网络抖动，客户端可ZK集群的session连接断了，那么zk以为客户端挂了，就会删除临时节点，这时候其他客户端就可以获取到分布式锁了，就可能产生并发问题。这个问题不常见是因为zk有重试机制，一旦zk集群检测不到客户端的心跳，就会重试，Curator客户端支持多种重试策略。多次重试之后还不行的话才会删除临时节点。（所以，选择一个合适的重试策略也比较重要，要在锁的粒度和并发之间找一个平衡。）

使用Zookeeper实现分布式锁的优点

1、有效的解决单点问题，不可重入问题，非阻塞问题以及锁无法释放的问题。实现起来较为简单。

使用Zookeeper实现分布式锁的缺点

2、性能上不如使用缓存实现分布式锁。 需要对ZK的原理有所了解。

分布式写锁
1.create一个PERSISTENT类型的znode，/Locks/write_lock
2.客户端创建SEQUENCE|EPHEMERAL类型的znode，名字是lockid开头，创建的znode是/Locks/write_lock/lockid0000000001
3.调用getChildren()不要设置Watcher获取/Locks/write_lock下的znode列表
4.判断自己步骤2创建znode是不是znode列表中最小的一个，如果是就代表获得了锁，如果不是往下走,调用exists()判断步骤2自己创建的节点编号小1的znode节点(也就是获取的znode节点列表中最小的znode)，并且设置Watcher，如果exists()返回false，执行步骤3,如果exists()返回true,那么等待zk通知，从而在回掉函数里返回执行步骤3

通过docker在Mac环境安装单机zookeeper

1

docker run --net=host -p 2181:2181,2888:2888,3888:3888 --name zoo00 -d zookeeper:latest

1
2
3
4
5
6
7
8

# 一般语法
./zkCli.sh -timeout 0 -r -server ip:port

# 连接到主机h1 超时时间3秒
./zkCli.sh -timeout 3000 -server h1:2181

# 有类似如下的命令提示符就表示连接成功了
[zk: h1:2181(CONNECTED) 0]

1
2

##docker进入客户端命令行模式
docker run -it --name my_zookeeper_1:zookeeper zookeeper zkCli.sh -server zookeeper

1
2
3

##进入zk内部
##需要exec进入Docker容器配置myid和hosts文件
docker exec -it ${容器id} /bin/bash

1
2

$ docker port ${容器id} 5000
$ 127.0.0.1:49155

通过docker在Mac环境安装集群zookeeper

1
2
3
4
5

## 首次执行，创建一个zk集群
COMPOSE_PROJECT_NAME=zk_test docker-compose up -d
##  以后运行，可以通过start\stop 控制启停
COMPOSE_PROJECT_NAME=zk_test docker-compose start
COMPOSE_PROJECT_NAME=zk_test docker-compose stop

1
2
3
4
5
6

$ COMPOSE_PROJECT_NAME=zk_test docker-compose ps
Name              Command               State           Ports
----------------------------------------------------------------------
zoo1   /docker-entrypoint.sh zkSe ...   Up      0.0.0.0:2181->2181/tcp
zoo2   /docker-entrypoint.sh zkSe ...   Up      0.0.0.0:2182->2181/tcp
zoo3   /docker-entrypoint.sh zkSe ...   Up      0.0.0.0:2183->2181/tcp

1
2
3
4
5
6

docker run -it --rm \
        --link zoo1:zk1 \
        --link zoo2:zk2 \
        --link zoo3:zk3 \
        --net zktest_default \
        zookeeper zkCli.sh -server zk1:2181,zk2:2181,zk3:2181

通过本地主机连接ZK集群
因为我们分别将zoo1, zoo2, zoo3 的2181端口映射到了 本地主机的2181, 2182, 2183端口上, 因此我们使用如下命令即可连接ZK集群:

1

zkCli.sh -server localhost:2181,localhost:2182,localhost:2183

zookeeper源码解析

客户端部分

1
2
3
4
5

WatchManager
ClientCnxn
ClientCnxnSocket
ClientCnxnSocketNetty/ClientCnxnSocketNIO
StaticHostProvider

Zookeeper考虑到第一次连接的时候，使用 StaticHostProvider#resolveAndShuffle方法进行一次 shuffle，避免第一台节点处于热点状态；而 Zookeeper的 Session连接断开之后，会使用 StaticHostProvider#next方法，从第一个连接地址开始逐个尝试；另外，在集群扩容/缩容的时候，使用 StaticHostProvider#updateServerList方法，更新服务器列表，并计算集群扩容的概率，对连接做重新分配，使得集群的负载更加均衡

服务端部分

写数据流程

关键类

1
2
3
4
5

NIOServerCnxn/NettyServerCnxn
FollowerRequestProcessor | ObserverRequestProcessor
PrepRequestProcessor
SyncRequestProcessor
FinalRequestProcessor

Leader处理写请求

按照ZooKeeperServer的文档中所示，事务处理的大体流程链应该为

PrepRequestProcessor -> SyncRequestProcessor -> FinalRequestProcessor

(1) 交付给FinalRequestProcessor处理器。FinalRequestProcessor处理器检查outstandingChanges队列中请求的有效性，若发现这些请求已经落后于当前正在处理的请求，那么直接从outstandingChanges队列中移除。

(2) 事务应用。之前的请求处理仅仅将事务请求记录到了事务日志中，而内存数据库中的状态尚未改变，因此，需要将事务变更应用到内存数据库。

(3) 将事务请求放入队列commitProposal。完成事务应用后，则将该请求放入commitProposal队列中，commitProposal用来保存最近被提交的事务请求，以便集群间机器进行数据的快速同步。

1
2
3
4
5
6
7
8

//PrepRequestProcessor

//FIFO 先入先出队列 最大容量Integer.MAX_VALUE
LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();

public void processRequest(Request request) {
    submittedRequests.add(request);
}

1
2
3
4
5
6
7
8
9

//SyncRequestProcessor

//FIFO 先入先出队列 最大容量Integer.MAX_VALUE
private final LinkedBlockingQueue<Request> queuedRequests = new LinkedBlockingQueue<Request>();

public void processRequest(Request request) {
       // request.addRQRec(">sync");
       queuedRequests.add(request);
   }

1.从头开始遍历outstandingChanges队列，移除所有事务id小于当前请求的事务id（这些是已经处理过的过期的ChangeRecord）；
2.从Request中获取请求头和事务，处理事务（下边详细说）；
3.如果是一个事务请求，执行addCommittedProposal（添加一个已提交的提案）方法（下边详细说）。
.更改节点；
2.更新最新事务id。这里就不贴代码了

这个方法主要是向committedLog里插入一条操作日志。committedLog是一个存在于内存中的事务日志队列：
protectedLinkedList committedLog = newLinkedList();
存入过程很简单，这里就不分析了，需要说明一点的是：如果超过的队列限制，会移除第一个，然后追加在最后。

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

// 位置:ZKDatabase
protected LinkedList<Proposal> committedLog = new LinkedList<Proposal>();

public void addCommittedProposal(Request request) {
    WriteLock wl = logLock.writeLock();
    try {
       wl.lock();
       if (committedLog.size() > commitLogCount) {
           committedLog.removeFirst();
           minCommittedLog = committedLog.getFirst().packet.getZxid();
       }
       if (committedLog.size() == 0) {
           minCommittedLog = request.zxid;
           maxCommittedLog = request.zxid;
       }
       ByteArrayOutputStream baos = new ByteArrayOutputStream();
       BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
       try {
           request.hdr.serialize(boa, "hdr");
           if (request.txn != null) {
               request.txn.serialize(boa, "txn");
           }
           baos.close();
       } catch (IOException e) {
         LOG.error("This really should be impossible", e);
     }
     QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, baos.toByteArray(), null);
     Proposal p = new Proposal();
     p.packet = pp;
     p.request = request;
     committedLog.add(p);
     maxCommittedLog = p.packet.getZxid();
    } finally {
        wl.unlock();
    }
}

SyncRequestProcessor的log和FinalRequestProcessor的log的区别在哪里
这是一个很容易弄混的问题，在这里特别说明一下：
SyncRequestProcessor的log是一个FileTxnSnapLog，这个log用于定期向磁盘刷写数据。
这个log队列记录的事务编号一般来说会大于快照记录的事务编号，所以在服务器启动时，需要从快照和这个log两部分读取恢复数据。
FinalRequestProcessor的log是一个LinkedList，它只存在于内存中，不会向磁盘刷写数据。并且它有容量限制，每次插入最新的，移除最老的（被移除的就真的被移除了，并不会像FileTxnSnapLog一样写入到磁盘上）。
它的作用是当一个事务被通过时，Zookeeper集群里的其他Follower学习时，从这里获取数据。

返回响应
最后，服务器会更新一下处理时间，状态，然后通过ServerCnxn.sendResponse方法将响应发送给请求方，至此，一次事务请求的处理流程结束。

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

long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
ReplyHeader hdr = new ReplyHeader(request.cxid, lastZxid, err.intValue());//序列化
zks.serverStats().updateLatency(request.createTime);
cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp, request.createTime, System.currentTimeMillis());
try {
    // NIO 操作返回网络请求
    cnxn.sendResponse(hdr, rsp, "response");
    if (closeSession) {
        cnxn.sendCloseSession();
    }
} catch (IOException e) {
    LOG.error("FIXMSG",e);
}

Follower、Observer处理写请求

为了事务的一致性，所有 Follower、Observer接收到的事务请求，都会通过 Learner#request方法，将请求发送给Leader节点来处理

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

// org.apache.zookeeper.server.quorum.FollowerRequestProcessor | ObserverRequestProcessor
switch (request.type) {
    case OpCode.sync:
        zks.pendingSyncs.add(request);
        zks.getFollower().request(request);
        break;
    case OpCode.create:
    case OpCode.create2:
    case OpCode.createTTL:
    case OpCode.createContainer:
    case OpCode.delete:
    case OpCode.deleteContainer:
    case OpCode.setData:
    case OpCode.reconfig:
    case OpCode.setACL:
    case OpCode.multi:
    case OpCode.check:
        zks.getFollower().request(request);
        break;
    case OpCode.createSession:
    case OpCode.closeSession:
        // Don't forward local sessions to the leader.
        if (!request.isLocalSession()) {
            zks.getFollower().request(request);
        }
        break;
}

存储部分

1

ZKDatabase

监控部分

1

OSMXBean

选举部分

1
2

QuorumPeer
FastLeaderElection

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

public Vote lookForLeader() throws InterruptedException {
    /*
     * Loop in which we exchange notifications until we find a leader
     */
    while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
        /*
         * Sends more notifications if have not received enough.
         * Otherwise processes new notification.
         */
        if(n == null) { // ... }
        else if (self.getCurrentAndNextConfigVoters().contains(n.sid)) {
            /*
             * Only proceed if the vote comes from a replica in the current or next voting view.
             */
            switch (n.state) {
            case LOOKING:
                if (getInitLastLoggedZxid() == -1) {
                    LOG.debug("Ignoring notification as our zxid is -1"); break;
                }
                if (n.zxid == -1) {
                    LOG.debug("Ignoring notification from member with -1 zxid" + n.sid); break;
                }
                // If notification > current, replace and send messages out
                if (n.electionEpoch > logicalclock.get()) {
                    logicalclock.set(n.electionEpoch);
                    recvset.clear();
                    if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                        updateProposal(n.leader, n.zxid, n.peerEpoch);
                    } else {
                        updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
                    }
                    sendNotifications();
                } else if (n.electionEpoch < logicalclock.get()) {
                    if(LOG.isDebugEnabled()){
                        LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
                                + Long.toHexString(n.electionEpoch)
                                + ", logicalclock=0x" + Long.toHexString(logicalclock.get()));
                    }
                    break;
                } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
                    updateProposal(n.leader, n.zxid, n.peerEpoch);
                    sendNotifications();
                }
                recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch)); // ...
            case OBSERVING: // ...
            case FOLLOWING: // ...
            case LEADING: // ...
            default: // ...
            }
        }
    }
}

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
    if (self.getQuorumVerifier().getWeight(newId) == 0) return false;
    /*
     * We return true if one of the following three cases hold:
     * 1- New epoch is higher
     * 2- New epoch is the same as current epoch, but new zxid is higher
     * 3- New epoch is the same as current epoch, new zxid is the same as current zxid, but server id is higher.
     */
    return ((newEpoch > curEpoch) ||
            ((newEpoch == curEpoch) &&
            ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
}

在QuorumPeer启动时，服务端节点节点状态是初始状态LOOKING。随后通过选举可能会有如下的状态。

zookeeper 相关问题

1.zookeeper数据同步原理

读取的时候不需要数据同步，写操作时follower的FollowerRequestProcessor会将该操作作为LEADER.REQEST发给leader。leader发起投票，由followers进行投票，leader对投票结果进行计算决定是否通过该决议，如果通过执行该决议（事务），否则什么也不做。

2.在Docker环境下，通常虚机会比较多，我们发现ZooKeeper不能承受太多节点。我们的游戏平台是一个多租户场景，需要频繁进行创建及删除，在这种情况下如果超过3000虚拟机，ZooKeeper就不行了。
当ZooKeeper挂了后，meta数据也会跟着丢了。Meta数据和和监控功能在ZooKeeper当初设计时并没有考虑的问题，因此后来我们打算自己造轮子，将这个需求实现。我也并非说ZooKeeper一无是处，ZooKeeper有它适合使用的场景，比如Hadoop那种场景ZooKeeper就可以工作得很好，但并不是所有场景都适用。Tim说过一句经典的话，Redis是把好锤子，但不能把所有存储的问题都当做钉子。ZooKeeper也是一样，不是所有的配置的问题都适合拿它来解决。

3.同意ZooKeeper是有很多运维的问题，一个解决方法是自己实现一个single node all in memory的lock service，然后运行在多个机器上，用ZooKeeper当作一个distributed lock来选一个master，这样ZooKeeper上的压力就小多了，而且不会随着集群大小增加而增加。

4.Kafka里使用ZooKeeper的方式有好几个地方可以借鉴。除了上面说的选一个master处理的方式外，为了避免订阅大量节点，也可以单设置一个变更节点，然后只订阅这个变更节点。
Kafka集群依赖于ZooKeeper，但ZooKeeper也是Kafka的瓶颈。

5.随着部署规模增大，客户端增多，ZooKeeper服务器中节点数量大增，核心ZooKeeper集群5台，每台服务器15000左右的长链接，近43万个节点，平均30余万watch。每个上线日，ZooKeeper服务器的流量都能上200mb/s，通过mntr的观察zk_outstanding_requests经常达到300M以上;

6.有业务方报告，应用启动后连上ZooKeeper但一直读失败(读数据返回为null)，也就是取服务列表失败，一段时间如半小时后自动恢复，查看zk server端日志，只看到一堆session过期的警告，没有其他异常;

7.由于历史原因，ZooKeeper服务器与其他应用共用，40万+节点中60%是我们的，其他应用也是重度依赖zookeeper，互相之间有影响;

8.ZooKeeper推送的频率、内容，我们自己不能控制，比如一个600 provider，12000 consumer的服务，如果新增一个provider，其实并不是每个consumer都需要通知的，但目前的机制下consumer监听一个目录，每个consumer都会得到一次provider列表推送;还有我们想在推送前根据一定的规则对provider列表做动态过滤排序，这个需求在zookeeper服务端也没法实现。

9.ZooKeeper客户端使用的是 172.17.xx.yy,172.17.xx.yz:2181 这样的ip串方式连接，后期想添加服务器分散主集群读写压力，也不好实现，因为需要更改地址串，需要update配置的客户端太多了;

10.跨机房容灾方面，一旦出现机房间通信问题，另一个机房的部署的Observer节点，就不可读写了;

11.provider约6万实例，consumer22万实例;
基于以上的原因，新的方案实现也很简单，自己实现了一个服务用来做注册中心的，实现的是服务注册/订阅方面的功能。它负责接收长链接并推送，数据存储在MySQL。

12.使用ZooKeeper主要是它实现了强一致性，你这个注册中心是单点的吧?master election用ZooKeeper?注册中心会成为新的瓶颈和故障点?
性能方面完全够用，因为这个服务也是集群部署的，客户端首先访问一个http接口拿到所有服务器的地址，并优先访问本机房的注册中心。注册中心基本读多写少，内存里面也缓存provider列表，不会有太大压力到MySQL;存储上通过MySQL实现一致性保证。

13.跨机房容灾方面有什么优化吗?如果网络没有足够的redundency，一旦出现network partiton，那么有一个机房就没办法到quorum了?
通过多机房部署，每个机房保证至少2个节点，本机房挂了还可以访问其它机房，并且在客户端本地还有缓存文件。

14.当配置变化，client以http轮询方式去感知吗?
Session过期那个，开始我们设的是很短，后来发现不对，网络一闪断，大量掉节点。现在我们自己实现的服务，半分钟一心跳，几个心跳没收到我才认为下线了。如果provider列表发生变化，我这边服务端会主动推送的，因为用的是TCP长连接。

15.Client有没有主动发现本地cache与服务端数据不一致的机制?例如当provider列表发生变化且服务端通过TCP长连给client推送失败的场景。
有，我们心跳带本地版本号的，如果与服务端不一致的话，会触发服务端再次推送;

16.而另外一知名大公司架构师认为如果consumer数量没有那么多，用ZooKeeper也能较好满足要求。其业务场景的ZooKeepr使用及运行情况如下。
线上的单个ZooKeeper节点平均连接数是6K，watcher是30万，netIO不高，ZooKeeper的机器配置比较差，用的是4G 4CPU 的虚拟机，5个节点，最多的consumer不超过200，平均consumer数量就5，6个左右。
目前线上provider数3万左右实例数，consumer数10万左右，目前运行的没什么问题。3万个，如果每个服务10个实例，也就3,000个服务，全公司的，也没多少。看起来是多了点。 跟公司发展历史及使用场景关系很大。
之前关于ZooKeeper踩坑最多的是在客户端上，最开始用的是netflix的curator(后来贡献给apache，但是我们用的是老的，没升级)，遇到网络闪断重连不上，然后死循环一样，升级到apache curator就好了。
后来遇到一个问题是，如果注册watcher太多，发生重连的时候，zk client会自动注册之前的所有watcher，这样会导致一个包的大小超过1M，然后也就重连不上又不断地重连，后来hack了zk client解决了这个问题，这个bug到目前为止zk官方仍然没有修复。
不过这些坑基本上遇到一个可以解决一个，但是目前有一个问题目前也没找到好的解决方案，那就是业务方系统load变高，或者发生长时间gc，导致zk重连甚至session过期。
另外关于zk连接，我们在最底下做了连接共享，因为好多服务都依赖zk，这个也降低了不少连接数。

17.客户端对ServerList的轮询机制是什么
随机，客户端在初始化( new ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) )的过程中，将所有Server保存在一个List中，然后随机打散，形成一个环。之后从0号位开始一个一个使用。
两个注意点：1. Server地址能够重复配置，这样能够弥补客户端无法设置Server权重的缺陷，但是也会加大风险。（比如: 192.168.1.1:2181,192.168.1.1:2181,192.168.1.2:2181). 2. 如果客户端在进行Server切换过程中耗时过长，那么将会收到SESSION_EXPIRED. 这也是上面第1点中的加大风险之处。

18.客户端如何正确处理CONNECTIONLOSS(连接断开) 和 SESSIONEXPIRED(Session 过期)两类连接异常
在ZooKeeper中，服务器和客户端之间维持的是一个长连接，在 SESSION_TIMEOUT 时间内，服务器会确定客户端是否正常连接(客户端会定时向服务器发送heart_beat),服务器重置下次SESSION_TIMEOUT时间。因此，在正常情况下，Session一直有效，并且zk集群所有机器上都保存这个Session信息。在出现问题情况下，客户端与服务器之间连接断了（客户端所连接的那台zk机器挂了，或是其它原因的网络闪断），这个时候客户端会主动在地址列表（初始化的时候传入构造方法的那个参数connectString）中选择新的地址进行连接。
好了，上面基本就是服务器与客户端之间维持长连接的过程了。在这个过程中，用户可能会看到两类客异常CONNECTIONLOSS(连接断开) 和SESSIONEXPIRED(Session 过期)。
CONNECTIONLOSS发生在上面红色文字部分，应用在进行操作A时，发生了CONNECTIONLOSS，此时用户不需要关心我的会话是否可用，应用所要做的就是等待客户端帮我们自动连接上新的zk机器，一旦成功连接上新的zk机器后，确认刚刚的操作A是否执行成功了。
SESSIONEXPIRED发生在上面蓝色文字部分，这个通常是zk客户端与服务器的连接断了，试图连接上新的zk机器，这个过程如果耗时过长，超过 SESSION_TIMEOUT 后还没有成功连接上服务器，那么服务器认为这个session已经结束了（服务器无法确认是因为其它异常原因还是客户端主动结束会话），开始清除和这个会话有关的信息，包括这个会话创建的临时节点和注册的Watcher。在这之后，客户端重新连接上了服务器在，但是很不幸，服务器会告诉客户端SESSIONEXPIRED。此时客户端要做的事情就看应用的复杂情况了，总之，要重新实例zookeeper对象，重新操作所有临时数据（包括临时节点和注册Watcher）。

19.不同的客户端对同一个节点是否能获取相同的数据,一个客户端修改了某个节点的数据，其它客户端能够马上获取到这个最新数据吗?
ZooKeeper不能确保任何客户端能够获取（即Read Request）到一样的数据，除非客户端自己要求：方法是客户端在获取数据之前调用org.apache.zookeeper.AsyncCallback.VoidCallback, java.lang.Object) sync.
通常情况下（这里所说的通常情况满足：1. 对获取的数据是否是最新版本不敏感，2. 一个客户端修改了数据，其它客户端需要不需要立即能够获取最新），可以不关心这点。
在其它情况下，最清晰的场景是这样：ZK客户端A对 /my_test 的内容从 v1->v2, 但是ZK客户端B对 /my_test 的内容获取，依然得到的是 v1. 请注意，这个是实际存在的现象，当然延时很短。解决的方法是客户端B先调用 sync(), 再调用 getData().

20.ZK为什么不提供一个永久性的Watcher注册机制
不支持用持久Watcher的原因很简单，ZK无法保证性能。

21.使用watch需要注意的几点
a. Watches通知是一次性的，必须重复注册.
b. 发生CONNECTIONLOSS之后，只要在session_timeout之内再次连接上（即不发生SESSIONEXPIRED），那么这个连接注册的watches依然在。
c. 节点数据的版本变化会触发NodeDataChanged，注意，这里特意说明了是版本变化。存在这样的情况，只要成功执行了setData()方法，无论内容是否和之前一致，都会触发NodeDataChanged。
d. 对某个节点注册了watch，但是节点被删除了，那么注册在这个节点上的watches都会被移除。
e. 同一个zk客户端对某一个节点注册相同的watch，只会收到一次通知。即

for( int i = 0; i < 3; i++ ){
zk.getData( path, true, null );
zk.getChildren( path, true );
}

22.我能否收到每次节点变化的通知？
如果节点数据的更新频率很高的话，不能。原因在于：当一次数据修改，通知客户端，客户端再次注册watch，在这个过程中，可能数据已经发生了许多次数据修改，因此，千万不要做这样的测试：”数据被修改了n次，一定会收到n次通知”来测试server是否正常工作。（我曾经就做过这样的傻事，发现Server一直工作不正常？其实不是）。即使你使用了GitHub上这个客户端也一样。

23.能为临时节点创建子节点吗?
不能。

24.是否可以拒绝单个IP对ZK的访问,操作
ZK本身不提供这样的功能，它仅仅提供了对单个IP的连接数的限制。你可以通过修改iptables来实现对单个ip的限制，当然，你也可以通过这样的方式来解决。https://issues.apache.org/jira/browse/ZOOKEEPER-1320

25.在getChildren(String path, boolean watch)这个API中，是注册了对节点子节点的变化，那么子节点的子节点变化能通知吗
不能

26.创建的临时节点什么时候会被删除，是连接一断就删除吗？延时是多少？
连接断了之后，ZK不会马上移除临时数据，只有当SESSIONEXPIRED之后，才会把这个会话建立的临时数据移除。因此，用户需要谨慎设置Session_TimeOut

27.zookeeper是否支持动态进行机器扩容？
如果目前不支持，那么要如何扩容呢？
截止2012-03-15，3.4.3版本的zookeeper，还不支持这个功能，在3.5.0版本开始，支持动态加机器了，期待下吧: https://issues.apache.org/jira/browse/ZOOKEEPER-107
目前只能通过修改zoo.cfg配置文件，然后逐台重启机器来实现扩容

28.echo stat | nc localhost $port
This ZooKeeper instance is not currently serving requests
这是最常见的问题了，一般可能有以下原因。
1、数据正在同步，需要等几分钟再试
2、可用节点数少于(n/2 + 1)，剩余的节点也无法服务（比如三个节点的zk集群，剩下一个节点，就无法服务了）
3、myid的数据错误了，需要zoo.cfg中的server.1(server.2,server.3)保持一致
4、java的通信端口不对，检查下zoo.cfg中serverlist中的端口是否每个机器一致，单机多部署时，查看是否和其他集群冲突

29.启动失败，进程退出
1、配置的log目录不存在，或者权限不够

zookeeper动态更换机器 更换zk节点

假设：目前有ip1，ip2，ip3机器，ip3坏掉了，需要更换为ip4
1、在ip4安装zk，同时配置文件和ip1完全一致（如果你拷贝的ip1的整个目录，记得清理数据目录）
2、修改ip4上myid文件，id号和坏掉的机器一致
3、修改ip4上zoo.cfg文件，把server.3修改为ip4
4、启动ip4的zk服务，这个时候ip4上不会同步任何数据
5、修改ip1的zoo.cfg文件，把server.3修改为ip4，重启。重启后，你仍然看不到ip4上有任何数据同步过来。
6、修改ip2的zoo.cfg文件，把server.3修改为ip4，重启。重启后，你现在可以看到data目录和log都有数据同步过来，一般同步几分钟就完成了，使用echo stat | nc localhost $port可以检查下。

其它资料

zk资料1

zk资料2

zk资料3

zk资料4

zk资料5

zk资料6

zk资料7

zk资料8

4.0里不再有了messageReceived或writeRequested处理器方法。

它们被inboundBufferUpdated和flush代替了。

用户的入队一个或多个消息到一个入站（或出站）缓存同时会出发一个inboundBUfferUpdated（或flush）事件。

messageReceived – inboundBufferUpdated

writeRequested – inboundBUfferUpdated（或flush）

SimpleChannelHandler – SimpleChannelInboundHandler

但是NIO并不是严格意义上的“异步IO”（Asynchronous），最大的原因就是Selector本身是阻塞的！！即selector需要通过线程阻塞的方式（其select方法）获取底层通道的事件变更，然后获取SelectionKey列表；那么对于“异步IO”（概念同JDK 7的AIO）在整个操作链路上均不需要任何阻塞（完全基于OS的IO事件），依赖基于事件驱动的Handler做数据处理。目前Netty尚没有集成AIO的相关特性，即Netty本身为非阻塞IO框

赏

谢谢你请我吃糖果

支付宝

微信