zookeeper

| 介绍

开源的分布式协调服务，雅虎创建，基于google chubby。

可以解决的问题

数据的发布/订阅（配置中心:disconf）
负载均衡（dubbo利用了zookeeper机制实现负载均衡）
命名服务
master选举(kafka、hadoop、hbase)
分布式队列
分布式锁。

特性

顺序一致性
原子性
可靠性
实时性

一旦一个事务被成功应用，客户端就能够立即从服务器端读取到事务变更后的最新数据状态；（zookeeper仅仅保证在一定时间内，近实时）

文件系统

数据模型和文件系统类似，每一个节点为称znode，zk中的最小数据单元，每个node上可以保存数据和挂载子节点。构成一个层次化的属性结构。

[外链图片转存失败(img-jI6h3nEF-1564466530214)(.assets/20141108213344_45.png)]

节点类型

持久化节点
持久化有序节点
临时节点

临时节点的生命期和会话周期相同

临时有序节点

存储数据大小：不要超过1M

命令

create [-s] [-e] path data

get path [watch]

set path data [version]

version表示锁的概念，乐观锁，数据库里面有一个version字段去控制数据的版本号

delete path [version]

必须从子节点开始删除，不会立即生效，有会话重试机制，过一段时间才会有

stat信息

名称	说明
cversion	子节点的版本号
dataVersion	数据的版本号
aclVersion	表示acl的版本号，修改节点权限
czxid	节点被创建时的事务ID
mzxid	节点最后一次被更新的事务ID
pzxid	当前节点下的子节点最后一次被修改时的事务ID
ctime
mtime
ephemeralOwner	创建临时节点时，会有一个sessionId
dataLength	数据长度

Watcher特性

分布式数据发布/订阅。zk允许客户端向服务器注册一个watcher监听，当服务器端的节点触发指定事件（数据改变、删除、子目录节点增加删除等）的时候会触发watcher，服务端会向客户端发送一个事件通知。

watcher的通知是一次性的，一量触发一次通知后，该watcher就失效。

ACL

提供控制节点访问权限的功能，用于有效的保证zk中数据的安全性，避免误操作而导致出现重大事故。

| 集群

角色类型

角色	说明
Leader	接收所有Follower的提案请求并统一协调发起投票，负责与所有的Follower进行内部的数据交换（同步）
Follower	直接为客户端服务并参与提案的投票，同时与Leader进行数据交换（同步）
Observer	直接为客户端服务但并不参与提案投票，同时也与Leader进行数据交换（同步）

follower不接收写请求

| zk一致性协议 - zab工作原理

leader选举

三种选主算法：leaderElection/AuthFastLeaderElection/FastLeaderElection（默认）

FastLeaderElection

serverid: 在配置server集群时，给定服务器的标识id（myid）
zxid: 64位Long类型，高32位（Epoch，选举轮数）表示当前属于那个leader统治，低32位递增的事务id号，zxid值越大，表示数据越新
server的状态：Looking，Following，Observering，Leading

选举流程

状态设置为LOOKING，初始化内部投票Vote(id, zxid)，将其广播到其它节点；首次投票都是自己作为Leader；然后循环等待其它节点的投票信息；
每收到一个Vote，都和自己的Vote数据PK，规则为ZXID大的优先，相等时给ID大的投票。若外部投票获胜，将该选票覆盖自己的Vote后再次广播出去；同时统计是否有过半的赞同者与自己的投票数据一致，无则继续等待Vote，有则需要判断Leader是否在赞同者之中，在则退出循环，选举结束，根据选举结果及各自角色切换状态。

每一次启动时初始化为Looking

[外链图片转存失败(img-oj3BwLk1-1564466530215)(.assets/20181129114824253-1564408184881.png)]

假设这些服务器从id1-5，依序启动：

因为一共5台服务器，只有超过半数以上，即最少启动3台服务器，集群才能正常工作。

（1）服务器1启动，发起一次选举。

服务器1投自己一票。此时服务器1票数一票，不够半数以上（3票），选举无法完成；
服务器1状态保持为LOOKING；

（2）服务器2启动，再发起一次选举。

服务器1和2分别投自己一票，此时服务器1发现服务器2的id比自己大，更改选票投给服务器2；
此时服务器1票数0票，服务器2票数2票，不够半数以上（3票），选举无法完成；
服务器1，2状态保持LOOKING；

（3）服务器3启动，发起一次选举。

与上面过程一样，服务器1和2先投自己一票，然后因为服务器3id最大，两者更改选票投给为服务器3；
此次投票结果：服务器1为0票，服务器2为0票，服务器3为3票。此时服务器3的票数已经超过半数（3票），服务器3当选Leader。
服务器1，2更改状态为FOLLOWING，服务器3更改状态为LEADING；

（4）服务器4启动，发起一次选举。

此时服务器1，2，3已经不是LOOKING状态，不会更改选票信息。交换选票信息结果：服务器3为3票，服务器4为1票。
此时服务器4服从多数，更改选票信息为服务器3；
服务器4并更改状态为FOLLOWING；

（5）服务器5启动，同4一样投票给3，此时服务器3一共5票，服务器5为0票；

服务器5并更改状态为FOLLOWING；

最终Leader是服务器3，状态为LEADING；

其余服务器是Follower，状态为FOLLOWING。

选主后的数据同步

二阶提交：leader生成提议并广播给followers，收到半数以上的ACK后，再广播commit消息，同时将事务操作应用到内存中。follower收到提议后先将事务写到本地事务日志，然后反馈ACK，等接到leader的commit消息时，才会将事务操作应用到内存中。

问题

假设一个事务P1在leader服务器被提交了，并且已经有过半的follower返回了ack。在leader节点把commit消息发送给folower机器之前leader服务器挂了怎么办?

新生成的Leader之前是follower，未收到commit消息，内存中是没有P1数据，ZAB协议保证选主后，P1是需要应用到集群中的。即通过选主后的数据同步来弥补。

已被处理的消息不能丢

消息在Leader上Commit了，但是其它Server还没有收到Commit消息已经挂了，为了实现已经被处理的消息不能丢，Zab使用了以下策略：

选举拥有zxid最大的节点作为新的leader；由于所有Proposal需要过半节点ACK，必须有一个节点保存了所有被COMMIT消息的Proposal状态；
新leader将自己事务日志中的proposal但未commit的消息处理；
新laeder与follower建立先进先出队列，先将自身有而follower没有的proposal发送给follower，再将这些proposal的commit命令发送给follower，以保证所有的 follower都保存了所有的proposal并已处理。

被丢弃的消息不能再次出现

场景：当leader接收到消息请求生成proposal后就挂了，其它follower并没有收到此proposal，重新选了leader后，这条消息是被跳过的。之前的leader重新启动成了follower，保留的被跳过的proposal状态，与整个系统状态不一致，需要删除。

Zab通过巧妙的设计zxid来实现这一目的，高32 epoch表示leader选举的轮数，每选一次，epoch+1，低32位是消息计数器，每接收到一条消息，这个值+1，新leader选举后这个值重置为0。这样，旧的leader挂了后重启，它不会被选举为leader，因为此时它的zxid肯定小于当前新的leader，当旧的leader作为follwer接入新leader后，新leader会让它将所有的拥有旧的epoch号的未被commit的proposal清除。

zab协议，一定需要保证已经被leader提交的事务也能够被所有follower提交

zab协议需要保证，在崩溃恢复过程中跳过哪些已经被丢弃的事务

事务操作

二阶提交，针对client的请求，leader服务器会为其生成对应的事务proposal，并将其发送给其它follower，然后收集各自的选票，最后进行事务提交。

[外链图片转存失败(img-rl2lu3Om-1564466530215)(.assets/1523632294541695.png)]

二阶提交过程中，移除了中断逻辑（事务回滚），所有follower要么正常反馈，要么抛弃。follower处理proposal后的处理很简单，写入事务日志，然后立马反馈ACK给leader，即是说如果不是网络，内存或磁盘问题，follower肯定会定入成功，并正常反馈ACK。Leader收到过半FollowerACK后，会广播Commit消息给所有learner，并将事务应用到内存；Learner收到commit消息后会将事务应用到内存，

什么情况下zab协议会进入崩溃恢复模式

当服务器启动时
当leader服务器出现网络中断、崩溃或者重启的情况
集群中已经不存在过半的服务器与该leader保持正常通信

zab协议进入崩溃恢复模式会做什么

当leader出现问题，zab协议进入崩溃恢复模式，并且选举出新的leader。当新的leader选举出来以后，如果集群中已经有过半机器完成了leader服务器的状态同（数据同步），退出崩溃恢复，进入消息广播模式
当新的机器加入到集群中的时候，如果已经存在leader服务器，那么新加入的服务器就会自觉进入数据恢复模式，找到leader进行数据同步

[外链图片转存失败(img-JUvIJkpS-1564466530215)(.assets/1564408409704.png)]

| 安装

单机安装

http://apache.fayea.com/zookeeper中下载

zkCli.sh -server ip:port

zoo.cfg配置说明

tickTime：时间单位，默认值是2000ms
initLimit：10*2000，leader服务器等待follow启动并完成同步的时间
syncLimit：5*2000，leader节点和follower节点进行心跳检测的最大延时时间
dataDir：zk服务器存储快照文件的目录
clientPort：客户端访问的端口
dataLogDir：表示配置zk事务日志的存储路径，默认在dataDir目录下

集群安装

[外链图片转存失败(img-tQ4V5Qtu-1564466530216)(.assets/1564197131345.png)]

第三步：启动每个zookeeper

如何增加observer节点

zoo.cfg中增加 peerType=observer

server.1=192.168.11.129:2888:3181
server.2=192.168.11.135:2888:3181
server.3=192.168.11.136:2888:3181:observer

三个端口：

2181：对Client端提供服务

2888：集群内机器通讯使用

3888：选举leader，leader挂掉时使用

使用docker安装，直接使用zookeeper:latest版本，最新版本为3.5.5。

使用docker-compose启动zk集群

官方安装文档

version: '3.1'services:zoo1:image: zookeeperrestart: alwayscontainer_name: zoo1ports:- 2181:2181environment:ZOO_MY_ID: 1ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181 server.2=zoo2:2888:3888;2181 server.3=zoo3:2888:3888;2181networks:zoo2:image: zookeeperrestart: alwayscontainer_name: zoo2ports:- 2182:2181environment:ZOO_MY_ID: 2ZOO_SERVERS: server.1=zoo1:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=zoo3:2888:3888;2181zoo3:image: zookeeperrestart: alwayscontainer_name: zoo3ports:- 2183:2181environment:ZOO_MY_ID: 3ZOO_SERVERS: server.1=zoo1:2888:3888;2181 server.2=zoo2:2888:3888;2181 server.3=0.0.0.0:2888:3888;2181

docker-compose up -d

[外链图片转存失败(img-zqNFwUNn-1564466530216)(.assets/1563270840302.png)]

分别将本地的2181、2182、2183端口映射到对应容器的2181端口上。

ZOO_MY_ID：表示zk服务的id

ZOO_SERVERS：表示zk集群的主机列表

查看各个节点的状态

[root@ceos03 zookeeper-svc]# docker exec -it zoo1 zkServer.sh status

ZooKeeper JMX enabled by default
Using config: /conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower

[root@ceos03 zookeeper-svc]# docker exec -it zoo2 zkServer.sh status

ZooKeeper JMX enabled by default
Using config: /conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower

[root@ceos03 zookeeper-svc]# docker exec -it zoo3 zkServer.sh status

ZooKeeper JMX enabled by default
Using config: /conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader

| 应用 - 分布式锁

锁分为两类，一类是保持独占，另一种是控制时序。

独占锁

所有client去创建同一个节点，如/Lock，最终成功创建的那个client拥有了这把锁，用完成之后再删除。

控制时序

所有client在/LOCKs下创建临时有序节点，编号最小的获得锁，用完删除，其它client依次使用，未获得锁的监控相临较小编号节点即可。

控制时序代码实现

导入jar包

查找zk包的版本，http://www.mvnrepository.com/search?q=zookeeper

<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.5.5</version>
</dependency>

zk连接

public class ZkClient {private final static String CONNECTSTRING="192.168.10.13:2181";private static int sessionTimeout=5000;//获取连接public static ZooKeeper getInstance() throws IOException, InterruptedException {final CountDownLatch conectStatus=new CountDownLatch(1);ZooKeeper zooKeeper=new ZooKeeper(CONNECTSTRING, sessionTimeout, new Watcher() {public void process(WatchedEvent event) {if(event.getState()== Event.KeeperState.SyncConnected){conectStatus.countDown();}}});conectStatus.await();return zooKeeper;}public static int getSessionTimeout() {return sessionTimeout;}
}

分布式锁实现

public class DistributeLock {private static final String ROOT_LOCKS="/LOCKS";//根节点private ZooKeeper zooKeeper;private int sessionTimeout; //会话超时时间private String lockID; //记录锁节点idprivate final static byte[] data={1,2}; //节点的数据private CountDownLatch countDownLatch=new CountDownLatch(1);public DistributeLock() throws IOException, InterruptedException {this.zooKeeper=ZookeeperClient.getInstance();this.sessionTimeout=ZookeeperClient.getSessionTimeout();}//获取锁的方法public boolean lock(){try {//1. create LOCKS/00000001lockID=zooKeeper.create(ROOT_LOCKS+"/",data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println(Thread.currentThread().getName()+"->成功创建了lock节点["+lockID+"], 开始去竞争锁");//2. get all nodeList<String> childrenNodes=zooKeeper.getChildren(ROOT_LOCKS,true);//获取根节点下的所有子节点//3. sort and get ministSortedSet<String> sortedSet=new TreeSet<String>();for(String children:childrenNodes){sortedSet.add(ROOT_LOCKS+"/"+children);}String first=sortedSet.first(); //拿到最小的节点if(lockID.equals(first)){//表示当前就是最小的节点System.out.println(Thread.currentThread().getName()+"->成功获得锁，lock节点为:["+lockID+"]");return true;}SortedSet<String> lessThanLockId=sortedSet.headSet(lockID);// 4. watch close less No. nodeif(!lessThanLockId.isEmpty()){String prevLockID=lessThanLockId.last();//拿到比当前LOCKID这个几点更小的上一个节点zooKeeper.exists(prevLockID,new LockWatcher(countDownLatch));countDownLatch.await(sessionTimeout, TimeUnit.MILLISECONDS);//上面这段代码意味着如果会话超时或者节点被删除（释放）了System.out.println(Thread.currentThread().getName()+" 成功获取锁：["+lockID+"]");}return true;} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}return false;}public boolean unlock(){System.out.println(Thread.currentThread().getName()+"->开始释放锁:["+lockID+"]");try {zooKeeper.delete(lockID,-1);System.out.println("节点["+lockID+"]成功被删除");return true;} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}return false;}
}

main

    public static void main(String[] args) {final CountDownLatch latch=new CountDownLatch(10);Random random=new Random();for(int i=0;i<10;i++){new Thread(()->{DistributeLock lock=null;try {lock=new DistributeLock();latch.countDown();latch.await();lock.lock();Thread.sleep(random.nextInt(500));} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}finally {if(lock!=null){lock.unlock();}}}).start();}}

| 应用 - master选举

master-slave模式

资源：https://www.cnblogs.com/sky-sql/p/6804467.html

[外链图片转存失败(img-6zPbMhnm-1564466530216)(.assets/393620-20160627154234109-1109968833.png)]

zookeeper进行master选举使用场景是什么？

分布式，Master往往用来协调集群中的其他系统单元，具有对分布式状态变更的决定权。如：Master负责处理一些复杂的逻辑，并将结果同步给集群中其他系统单元。

多个client同时去创建相同的节点，创建成功的即是master；

创建失败的需要获取节点上的数据，即具体的master信息；

代码实现

1. jar依赖

    <dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.5.5</version></dependency><dependency><groupId>com.101tec</groupId><artifactId>zkclient</artifactId><version>0.11</version></dependency>

2. MasterSelector

package com.wjg.master_slave;import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;public class MasterSelector {private ZkClient zkClient;private final static String MASTER_PATH = "/master"; //需要争抢的节点private IZkDataListener dataListener; //注册节点内容变化private UserCenter server;  //其他服务器private UserCenter master;  //master节点private boolean isRunning = false;ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);public MasterSelector(UserCenter server, ZkClient zkClient) {System.out.println("[" + server + "] 去争抢master权限");this.server = server;this.zkClient = zkClient;this.dataListener = new IZkDataListener() {@Overridepublic void handleDataChange(String s, Object o) throws Exception {}@Overridepublic void handleDataDeleted(String s) throws Exception {//节点如果被删除, 发起选主操作，这里可以处理，判断之前的master是否为本机，若是，则立即进行选举，否则 ，等待5s再进行选举if(master.getMc_id() != server.getMc_id())TimeUnit.SECONDS.sleep(4);chooseMaster();}};}public void start() {//开始选举if (!isRunning) {isRunning = true;zkClient.subscribeDataChanges(MASTER_PATH, dataListener); //注册节点事件chooseMaster();}}public void stop() {//停止if (isRunning) {isRunning = false;scheduledExecutorService.shutdown();zkClient.unsubscribeDataChanges(MASTER_PATH, dataListener);releaseMaster();}}//具体选master的实现逻辑private void chooseMaster() {if (!isRunning) {System.out.println("当前服务没有启动");return;}try {zkClient.createEphemeral(MASTER_PATH, server);master = server; //把server节点赋值给masterSystem.out.println(master + "->我现在已经是master，你们要听我的");//定时器//master释放(master 出现故障）,5秒后释放一次scheduledExecutorService.schedule(() -> {releaseMaster();//释放锁}, 2, TimeUnit.SECONDS);} catch (ZkNodeExistsException e) {//表示master已经存在UserCenter userCenter = zkClient.readData(MASTER_PATH, true);if (userCenter == null) {System.out.println("启动操作：");chooseMaster(); //再次获取master} else {master = userCenter;}}}private void releaseMaster() {//释放锁(故障模拟过程)//判断当前是不是master，只有master才需要释放if (checkIsMaster()) {System.out.println(server+" release master!");zkClient.delete(MASTER_PATH); //删除}}private boolean checkIsMaster() {//判断当前的server是不是masterUserCenter userCenter = zkClient.readData(MASTER_PATH);if (userCenter.getMc_name().equals(server.getMc_name())) {master = userCenter;return true;}return false;}}

3. UserCenter

package com.wjg.master_slave;import java.io.Serializable;public class UserCenter implements Serializable {private static final long serialVersionUID = -1776114173857775665L;private int mc_id; //机器信息private String mc_name;//机器名称public int getMc_id() {return mc_id;}public void setMc_id(int mc_id) {this.mc_id = mc_id;}public String getMc_name() {return mc_name;}public void setMc_name(String mc_name) {this.mc_name = mc_name;}@Overridepublic String toString() {return "UserCenter{" +"mc_id=" + mc_id +", mc_name='" + mc_name + '\'' +'}';}
}

4. main

public class App
{private final static String CONNECTSTRING="192.168.10.13:2181";public static void main(String[] args) throws IOException {List<MasterSelector> selectorLists = new ArrayList<>();for (int i = 0; i < 10; i++) {int id = i;Thread th = new Thread(() -> {System.out.println(id+" started!");ZkClient zkClient = new ZkClient(CONNECTSTRING, 5000,10000,new SerializableSerializer());UserCenter userCenter = new UserCenter();userCenter.setMc_id(id);userCenter.setMc_name("客户端：" + id);MasterSelector selector = new MasterSelector(userCenter, zkClient);selectorLists.add(selector);selector.start();//触发选举操作});th.start();}System.out.println("press any key to stop!");System.in.read();for (MasterSelector selector : selectorLists) {selector.stop();}}
}

说明：在实际生产环境中，可能会由于插拔网线等导致网络短时的不稳定，也就是网络抖动。由于正式生产环境中可能server在zk上注册的信息是比较多的，而且server的数量也是比较多的，那么每一次切换主机，每台server要同步的数据量(比如要获取谁是master，当前有哪些salve等信息，具体视业务不同而定)也是比较大的。那么我们希望，这种短时间的网络抖动最好不要影响我们的系统稳定，也就是最好选出来的master还是原来的机器，那么就可以避免发现master更换后，各个salve因为要同步数据等导致的zk数据网络风暴。所以在抢主的时候，如果之前主机是本机，则立即抢主，否则延迟5s抢主。这样就给原来主机预留出一定时间让其在新一轮选主中占据优势，从而利于环境稳定。

| 应用 - 分布队列

先进先出队列

通过getchildren获取指定根节点下的所有子节点，子节点就是任务
确定自己节点在子节点中的顺序
如果自己不是最小的节点，那么监控比自己小的上一个子节点，否则处于等待
接收watcher通知，重复流程

zookeeper介绍相关推荐

zookeeper介绍及使用
zookeeper介绍及使用 zookeeper介绍什么是分布式协调技术`在这里插入代码片` 什么是分布式锁什么是zookeeper docker 安装 zookeeper zookeeper介绍 ...
Zookeeper介绍、原理及应用
Zookeeper简介 Zookeeper 分布式服务框架是 Apache Hadoop 的一个子项目,它主要是用来解决分布式应用中经常遇到的一些数据管理问题,如:统一命名服务.状态同步服务.集群管理 ...
zookeeper介绍及集群的搭建(利用虚拟机)
ZooKeeper ZooKeeper是一个分布式的,开放源码(apache)的分布式应用程序协调服务,是Google的Chubby一个开源的实现,是Hadoop和Hbase.dubbox.kaf ...
zookeeper 密码_Dubbo、ZooKeeper介绍
dubbo是一个分布式架构的服务框架,一般结合maven的模块式开发使用. 传统的单架构项目,不方便维护和升级: 通过maven的模块式开发,就可以把一个单架构的工程,拆封成一个一个的小模块,包括(j ...
Zookeeper介绍（通俗易懂）
Zookeeper简介 1.1 什么是Zookeeper ZooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,是Google的Chubby一个开源的实现,是大数据生态中的重要组件.它是 ...
zookeeper版本更新_zookeeper介绍及运维实践
Zookeeper介绍首先介绍下Zookeeper的背景.数据类型.使用场景以及ZAB协议,让大家对Zookeeper有一个清晰的认识. Zookeeper概述 ZooKeeper是一个分布式的.开 ...
docker集群——介绍Mesos+Zookeeper+Marathon的Docker管理平台
容器为用户打开了一扇通往新世界的大门,真正进入这个容器的世界后,却发现新的生态系统如此庞大.在生产使用中,不论个人还是企业,都会提出更复杂的需求.这时,我们需要众多跨主机的容器协同工作,需要支持各种类 ...
【Zookeeper入门】相关概念总结
1. 前言相信大家对 ZooKeeper 应该不算陌生.但是你真的了解 ZooKeeper 到底有啥用不?如果别人/面试官让你给他讲讲对于 ZooKeeper 的认识,你能回答到什么地步呢? 拿我自 ...
ZooKeeper基础学习
简介: ZooKeeper:为分布式应用提供了高效且可靠的分布式协调服务,提供了诸如统一命名服务.配置管理和分布式锁等分布式的基础服务. Zookeeper介绍: 是一个开放源代码的分布式协调服务,设 ...

zookeeper介绍