zookeeper源码解析

ZAB协议，原子广播协议

ZAB的核心思想

zxid

大量的服务的上线、注册、心跳的压力，达到了每秒几万，甚至上十万，zk的单个leader写入是扛不住那么大的压力的

读可以有每秒几万QPS

所有机器的配置文件，都要加入一个server.4=zk04:2888:3888:observer

哪怕客户端断开连接，也一直存在

客户端断开连接，节点就没了

创建节点的时候自增加全局递增的序号

临时顺序节

持久节点

临时节点

在分布式系统的协调中应用

分布式架构中的系统A监听一个数据的变化，如果分布式架构中的系统B更新了那个数据/节点，zk反过来通知系统A这个数据的变化

zk servers最多只能允许你的一台机器跟他建立60个连接

commit的时候，需要fsync到磁盘上去

但是，有可能会丢失部分os cache里没刷入磁盘的数据，如果是leader宕机

PathChildrenCache pathChildrenCache = new PathChildrenCache(
client, “/yh/config”, true);
pathChildrenCache.start();

pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
       public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
      }
});

cache就是把zk里的数据缓存到了你的客户端里来

针对这个缓存的数据加监听器，去观察zk里的数据的变化

final NodeCache nodeCache = new NodeCache(client, "/cluster");
nodeCache.start();

nodeCache.getListenable().addListener(new NodeCacheListener() {
              public void nodeChanged() throws Exception {
                 Stat stat = client.checkExists().forPath(“/yh/config”);
                 if(stat == null) { }
                 else { nodeCache.getCurrentData(); }
              }
});

final TreeCache treeCache = TreeCache.newBuilder(client, “/yh/config”).setCacheData(true).setMaxDepth(2)
.build();

增加监听：
treeCache.getListenable().addListener((curatorFramework, treeCacheEvent) -> {
//增加或修改
if(treeCacheEvent.getType().equals(Type.NODE_ADDED)||treeCacheEvent.getType().equals(Type.NODE_UPDATED)){
       String value = new String(treeCacheEvent.getData().getData());
       localCache.put(key, Optional.ofNullable(value));//加入本地缓存
}
//删除
if(treeCacheEvent.getType().equals(Type.NODE_REMOVED)){
        localCache.invalidate(key);//删除本地缓存
}
});
// 没有开启模式作为入参的方法
treeCache.start();
}

指定的目录下创建一个子节点，创建一个临时顺序节点

获取到的子节点做一个排序，然后看看自己是不是第一个子节点

  LeaderLatch leaderLatch = new LeaderLatch(client, "/leader/latch");
leaderLatch.start(); // 尝试竞争为leader
leaderLatch.await(); // 直到等待他成为leader再往后走，异步化，底层会调用norifyall唤醒这个行代码

分布式锁来竞争成为leader的，如果说你获取到了锁，就说明你是leader，获取锁也是跟第一种一样，创建临时顺序节点

LeaderSelector leaderSelector = new LeaderSelector(
    client,
    "/leader/election",
    new LeaderSelectorListener() {
             public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
             System.out.println("你已经成为了leader......");
             // 在这里干leader所有的事情，此时方法不能退出
             Thread.sleep(Integer.MAX_VALUE);
             }

            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                  System.out.println("连接状态的变化......");
                  if(connectionState.equals(ConnectionState.LOST)) {
                       throw new CancelLeadershipException();
                   }
            }
});

leaderSelector.start();

CuratorFramework

启动一个线程，网络连接监听

return new ZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly);

底层的源码，就会去跟zk的一台机器建立真正的TCP长连接

之后就会进行心跳，维护一个session，发送各种请求过去给zk server

注册watcher，zk有事件变更，会通过TCP长连接，反向通知你的ZooKeeper客户端，他会回调你提供给他的watcher

/locks/lock/__WRITE__{很多临时顺序节点}，排队等待加写锁

/locks/lock/__READ__{很多临时顺序节点}，加读锁

羊群效用

http://ant.apache.org/bindownload.cgi

下载apache-ant-1.10.5-bin.tar.gz

配置环境变量

https://zookeeper.apache.org/releases.html#download

每台机器都可以成为leader，可以干leader的事儿，每台机器也都可以成为follower

里面任何一个节点都可以认为是一个peer

zk的进程，就是QuorumPeerMain进程，java命令，通过java命令启动一个jvm进程

zk是Peer-to-Peer的集群架构，里面有leader-follower的角色区分，在多个peers节点之间要选举一个Leader出来的话

QuorumPeerMain启动

启动ZooKeeperServerMain类

达到了quorum的数量，此时就可以发起leader选举了

myid=0的机器，投票，（0, 0），第二个0代表zxid，发送给当前集群里其他的机器

myid=1的机器，投票，（1, 0），发送给当前集群里其他的机器

优先是zxid最大的机器成为leader

myid=0的机器，投出去（1,0），收到的票是（1,0）

票数已经达到了集群的quorum大多数了

选举就结束，确定（1,0）机器是leader

myid=2的机器，(2,10)

type，zxid

createSession（）

expirationInterval=2S间隔

（6/expirationInterval+1）*expirationInterval

达到效果：不同的expireTime得到相同的tickTime

分桶数据结构<expireTime,SessionSet>

session tracker而言就是不停的过期一个一个分桶

客户端是如何定期发送Ping心跳到服务端

服务端是如何接收和处理Ping心跳请求

1，返回ack给leader，会尽快的用while循环，把积压在内存队列里排队的proposal的请求全部都尽快的写入到磁盘上的事务日志里去

一旦将队列里积压的proposal都写入事务日志了，此时就可以执行flush了

2，如果你连续写入1000条数到事务日志里去，此时也会强制性执行flush以及后续的操作

1~10000条数据，在这个事务日志文件01里

10001~20000条数据，在这个事务日志文件02里

删除包含了1~10000条事务的数据快照

删除包含了1~10000条事务的日志

删除包含了10001~20000条事务的日志

对于zk来说，核心的数据结构，并不是文件目录树的结构

而是map接口，他所有操作都是针对一个path

所有watcher都是监听dataTree

create [-s] [-e] path data acl s表示临时 e表示序号递增

client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/demo01/world");

ls path [watch] 后面可以跟watch监听某个节点

byte[] bytes = client.getData().forPath("/demo01/world");

获取路径数据的同时，可以对这个路径加一个监听器

set path data [version]

client.setData().forPath("/demo01/world","您好啊".getBytes());

delete path [version]

client.delete().forPath("/hello5");

rmr path

判断当前节点是否存在

  //设置节点cache
TreeCache treeCache = new TreeCache(client,"/");

//设置监听器 
treeCache.getListenable().addListener(new TreeCacheListener() {
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData data = event.getData();
                if(data!=null){
                    //switch 判断各种可能性
                    switch (event.getType()){
                        case NODE_ADDED:
                        System.out.println("节点新增:路径:"+data.getPath()+"数据:"+new String(data.getData()));
                        break;
                        case NODE_REMOVED:
                        System.out.println("节点删除,路径:"+data.getPath()+"数据:"+new String(data.getData()));
                        break;
                        case NODE_UPDATED:
                        System.out.println("节点修改,路径:"+data.getPath()+"数据:"+new String(data.getData()));
                        break;
                    }
                }

写请求以字节流发送到leader

提供监听和回调的机制，只有把这个功能给实现了，此时zookeeper才是工业级的

三种监听器

在客户端也要保存一份；在zk服务端也要保存一份

什么时候在客户端完成注册

zk服务端处理查询请求中的watch标识

增删改事件发生的时候触发监听器

服务端触发监听器的时候，回调客户端连接

zk客户端收到watch监听通知的请求之后，如何处理

会删除掉你所有的这个客户端创建的临时节点

删除了之后到commit的时候，删除了内存里的节点，一定会触发对应的监听器的

客户端加在服务端上的这些监听器都给干掉

客户端必须去找其他的机器，leader，follower，去建立长连接

建立会话，重新把自己内存里注册的那些监听器，在新建立连接的机器上去进行一个重新的注册

三种宕机

客户端如何感知服务器宕机

follower宕机，leader是如何感知

Leader宕机，各个follower会如何感知

zk重启的时候，是如何加载磁盘上的数据进行恢复的