zookeeper --- 高级篇

一、zookeeper 事件监听机制

1.1、watcher概念

zookeeper提供了数据的发布/订阅功能，多个订阅者可同时监听某一特定主题对象，当该主题对象的自身状态发生变化时(例如节点内容改变、节点下的子节点列表改变等)，会实时、主动通知所有订阅者

zookeeper采用了Watcher机制实现数据的发布/订阅功能。该机制在被订阅对象发生变化时会异步通知客户端，因此客户端不必在Watcher注册后轮询阻塞，从而减轻了客户端压力。

zooKeeper 允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。

zooKeeper 原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便，需要开发人员自己反复注册Watcher，比较繁琐。

Curator引入了 Cache 来实现对 zooKeeper 服务端事件的监听。

watcher机制实际上与观察者模式类似，也可看作是一种观察者模式在分布式场景下的实现方式。

zooKeeper提供了三种Watcher：

  • NodeCache : 只是监听某一个特定的节点
  • PathChildrenCache : 监控一个ZNode的子节点.
  • TreeCache : 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合

1.2、watcher架构

Watcher实现由三个部分组成：

  • Zookeeper服务端
  • Zookeeper客户端
  • 客户端的ZKWatchManager对象

客户端首先将Watcher注册到服务端，同时将Watcher对象保存到客户端的Watch管
理器中。当ZooKeeper服务端监听的数据状态发生变化时，服务端会主动通知客户端，
接着客户端的Watch管理器会触发相关Watcher来回调相应处理逻辑，从而完成整体的数
据发布/订阅流程。

在这里插入图片描述

1.3、watcher特性

特性说明
一次性watcher是一次性的，一旦被触发就会移除，再次使用时需要重新注册
客户端顺序回调watcher回调是顺序串行化执行的，只有回调后客户端才能看到最新的数据状态。一个watcher回调逻辑不应该太多，以免影响别的watcher执行
轻量级WatchEvent是最小的通信单元，结构上只包含通知状态、事件类型和节点路径，并不会告诉数据节点变化前后的具体内容；
时效性watcher只有在当前session彻底失效时才会无效，若在session有效期内快速重连成功，则watcher依然存在，仍可接收到通知；

1.4、watcher接口设计

Watcher是一个接口，任何实现了Watcher接口的类就是一个新的Watcher。
Watcher内部包含了两个枚举类：KeeperState、EventType
在这里插入图片描述

  • Watcher通知状态(KeeperState)
    • KeeperState是客户端与服务端连接状态发生变化时对应的通知类型。路径为
      org.apache.zookeeper.Watcher.Event.KeeperState，是一个枚举类，其枚举属性
      如下：
枚举属性说明
SyncConnected客户端与服务器正常连接时
Disconnected客户端与服务器断开连接时
Expired会话session失效时
AuthFailed身份认证失败时
  • Watcher事件类型(EventType)
    • EventType是数据节点(znode)发生变化时对应的通知类型。EventType变化时
      KeeperState永远处于SyncConnected通知状态下；当KeeperState发生变化时，
      EventType永远为None。其路径为org.apache.zookeeper.Watcher.Event.EventType，
      是一个枚举类，枚举属性如下：
枚举属性说明
None无
NodeCreatedWatcher监听的数据节点被创建时
NodeDeletedWatcher监听的数据节点被删除时
NodeDataChangedWatcher监听的数据节点内容发生变更时(无论内容数据是否变化)
NodeChildrenChangedWatcher监听的数据节点的子节点列表发生变更时

注：客户端接收到的相关事件通知中只包含状态及类型等信息，不包括节点变化前后的具体内容，变化前的数据需业务自身存储，变化后的数据需调用get等方法重新获取；

1.5、捕获相应的事件

上面讲到zookeeper客户端连接的状态和zookeeper对znode节点监听的事件类型，下面我们来讲解如何建立zookeeper的watcher监听。在zookeeper中采用zk.getChildren(path, watch)、zk.exists(path, watch)、zk.getData(path, watcher, stat)这样的方式为某个znode注册监听。

下表以node-x节点为例，说明调用的注册方法和可监听事件间的关系：

注册方式CreatedChildrenChangedChangedDeleted
zk.exists(“/node-x”,watcher)可监控可监控可监控
zk.getData(“/node-x”,watcher)可监控可监控
zk.getChildren(“/node-x”,watcher)可监控可监控

1.6、注册watcher的方法

1.6.1、客服端与服务器的连接状态

KeeperState 通知状态
SyncConnected:客户端与服务器正常连接时
Disconnected:客户端与服务器断开连接时
Expired:会话session失效时
AuthFailed:身份认证失败时
事件类型为:None

1.6.2、检查节点是否存在

// 使用连接对象的监视器
exists(String path, boolean b)
// 自定义监视器
exists(String path, Watcher w)
// NodeCreated:节点创建
// NodeDeleted:节点删除
// NodeDataChanged:节点内容发生变化
  • path： znode路径。
  • b：是否使用连接对象中注册的监视器。
  • w：监视器对象。

1.6.3、查看节点

// 使用连接对象的监视器
getData(String path, boolean b, Stat stat)
// 自定义监视器
getData(String path, Watcher w, Stat stat)
// NodeDeleted:节点删除
// NodeDataChanged:节点内容发生变化
  • path： znode路径。
  • b：是否使用连接对象中注册的监视器。
  • w：监视器对象。
  • stat： 返回znode的元数据。

1.6.4、查看子节点

// 使用连接对象的监视器
getChildren(String path, boolean b)
// 自定义监视器
getChildren(String path, Watcher w)
// NodeChildrenChanged:子节点发生变化
// NodeDeleted:节点删除
  • path： znode路径。
  • b： 是否使用连接对象中注册的监视器。
  • w：监视器对象。

1.7、配置中心案例

工作中有这样的一个场景: 数据库用户名和密码信息放在一个配置文件中，应用读取该配置文件，配置文件信息放入缓存。

若数据库的用户名和密码改变时候，还需要重新加载缓存，比较麻烦，通过ZooKeeper可以轻松完成，当数据库发生变化时自动完成缓存同步。

设计思路：

  1. 连接zookeeper服务器
  2. 读取zookeeper中的配置信息，注册watcher监听器，存入本地变量
  3. 当zookeeper中的配置信息发生变化时，通过watcher的回调方法捕获数据变化事件
  4. 重新获取配置信息

案例:

public class MyConfigCenter implements Watcher {// zk的连接串String IP = "192.168.60.130:2181";// 计数器对象CountDownLatch countDownLatch = new CountDownLatch(1);// 连接对象static ZooKeeper zooKeeper;// 用于本地化存储配置信息private String url;private String username;private String password;@Overridepublic void process(WatchedEvent event) {try {// 捕获事件状态if (event.getType() == Event.EventType.None) {if (event.getState() == Event.KeeperState.SyncConnected){System.out.println("连接成功");countDownLatch.countDown();} else if (event.getState() ==Event.KeeperState.Disconnected) {System.out.println("连接断开!");} else if (event.getState() == Event.KeeperState.Expired){System.out.println("连接超时!");// 超时后服务器端已经将连接释放，需要重新连接服务器端zooKeeper = new ZooKeeper("192.168.60.130:2181",6000,new ZKConnectionWatcher());} else if (event.getState() ==Event.KeeperState.AuthFailed) {System.out.println("验证失败!");}// 当配置信息发生变化时} else if (event.getType() == EventType.NodeDataChanged) {initValue();}} catch (Exception ex) {ex.printStackTrace();}}// 构造方法public MyConfigCenter() {initValue();}// 连接zookeeper服务器，读取配置信息public void initValue() {try {// 创建连接对象zooKeeper = new ZooKeeper(IP, 5000, this);// 阻塞线程，等待连接的创建成功countDownLatch.await();// 读取配置信息this.url = new String(zooKeeper.getData("/config/url", true,null));this.username = new String(zooKeeper.getData("/config/username", true, null));this.password = new String(zooKeeper.getData("/config/password", true, null));} catch (Exception ex) {ex.printStackTrace();}}public static void main(String[] args) {try {MyConfigCenter myConfigCenter = new MyConfigCenter();for (int i = 1; i <= 20; i++) {Thread.sleep(5000);System.out.println("url:"+myConfigCenter.getUrl());System.out.println("username:"+myConfigCenter.getUsername());System.out.println("password:"+myConfigCenter.getPassword());System.out.println("########################################");}} catch (Exception ex) {ex.printStackTrace();}}public String getUrl() {return url;}public void setUrl(String url) {this.url = url;}public String getUsername() {return username;}public void setUsername(String username) {this.username = username;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;}
}

1.8、生成分布式唯一ID

在过去的单库单表型系统中，通常可以使用数据库字段自带的auto_increment
属性来自动为每条记录生成一个唯一的ID。但是分库分表后，就无法在依靠数据库的
auto_increment属性来唯一标识一条记录了。此时我们就可以用zookeeper在分布式环
境下生成全局唯一ID。

设计思路：

  1. 连接zookeeper服务器
  2. 指定路径生成临时有序节点
  3. 取序列号及为分布式环境下的唯一ID

案例:

public class GloballyUniqueId implements Watcher {// zk的连接串String IP = "192.168.60.130:2181";// 计数器对象CountDownLatch countDownLatch = new CountDownLatch(1);// 用户生成序号的节点String defaultPath = "/uniqueId";// 连接对象ZooKeeper zooKeeper;@Overridepublic void process(WatchedEvent event) {try {// 捕获事件状态if (event.getType() == Watcher.Event.EventType.None) {if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {System.out.println("连接成功");countDownLatch.countDown();} else if (event.getState() == Watcher.Event.KeeperState.Disconnected) {System.out.println("连接断开!");} else if (event.getState() == Watcher.Event.KeeperState.Expired) {System.out.println("连接超时!");// 超时后服务器端已经将连接释放，需要重新连接服务器端zooKeeper = new ZooKeeper(IP, 6000, new ZKConnectionWatcher());} else if (event.getState() == Watcher.Event.KeeperState.AuthFailed) {System.out.println("验证失败!");}}} catch (Exception ex) {ex.printStackTrace();}}// 构造方法public GloballyUniqueId() {try {//打开连接zooKeeper = new ZooKeeper(IP, 5000, this);// 阻塞线程，等待连接的创建成功countDownLatch.await();} catch (Exception ex) {ex.printStackTrace();}}// 生成id的方法public String getUniqueId() {String path = "";try {//创建临时有序节点path = zooKeeper.create(defaultPath, new byte[0],Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);} catch (Exception ex) {ex.printStackTrace();}// /uniqueId0000000001return path.substring(9);}public static void main(String[] args) {GloballyUniqueId globallyUniqueId = new GloballyUniqueId();for (int i = 1; i <= 5; i++) {String id = globallyUniqueId.getUniqueId();System.out.println(id);}}
}

1.9、分布式锁

在这里插入图片描述

分布式锁有多种实现方式，比如通过数据库、redis都可实现。作为分布式协同工具ZooKeeper，当然也有着标准的实现方式。下面介绍在zookeeper中如何实现排他锁。

设计思路：

  1. 每个客户端往/Locks下创建临时有序节点/Locks/Lock_000000001
  2. 客户端取得/Locks下子节点，并进行排序，判断排在最前面的是否为自己，如果自己的锁节点在第一位，代表获取锁成功
  3. 如果自己的锁节点不在第一位，则监听自己前一位的锁节点。例如，自己锁节点Lock_000000001
  4. 当前一位锁节点（Lock_000000002）的逻辑
  5. 监听客户端重新执行第2步逻辑，判断自己是否获得了锁

案例:

public class MyLock {// zk的连接串String IP = "192.168.60.130:2181";// 计数器对象CountDownLatch countDownLatch = new CountDownLatch(1);//ZooKeeper配置信息ZooKeeper zooKeeper;private static final String LOCK_ROOT_PATH = "/Locks";private static final String LOCK_NODE_NAME = "Lock_";private String lockPath;// 打开zookeeper连接public MyLock() {try {zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if (event.getType() == Event.EventType.None) {if (event.getState() ==Event.KeeperState.SyncConnected) {System.out.println("连接成功!");countDownLatch.countDown();}}}});countDownLatch.await();} catch (Exception ex) {ex.printStackTrace();}}//获取锁public void acquireLock() throws Exception {//创建锁节点createLock();//尝试获取锁attemptLock();}//创建锁节点private void createLock() throws Exception {//判断Locks是否存在，不存在创建Stat stat = zooKeeper.exists(LOCK_ROOT_PATH, false);if (stat == null) {zooKeeper.create(LOCK_ROOT_PATH, new byte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}// 创建临时有序节点lockPath = zooKeeper.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println("节点创建成功:" + lockPath);}//监视器对象，监视上一个节点是否被删除Watcher watcher = new Watcher() {@Overridepublic void process(WatchedEvent event) {if (event.getType() == Event.EventType.NodeDeleted) {synchronized (this) {notifyAll();}}}};//尝试获取锁private void attemptLock() throws Exception {// 获取Locks节点下的所有子节点List<String> list = zooKeeper.getChildren(LOCK_ROOT_PATH, false);// 对子节点进行排序Collections.sort(list);// /Locks/Lock_000000001int index = list.indexOf(lockPath.substring(LOCK_ROOT_PATH.length() + 1));if (index == 0) {System.out.println("获取锁成功!");return;} else {// 上一个节点的路径String path = list.get(index ‐ 1);Stat stat = zooKeeper.exists(LOCK_ROOT_PATH + "/" + path,watcher);if (stat == null) {attemptLock();} else {synchronized (watcher) {watcher.wait();}attemptLock();}}}//释放锁public void releaseLock() throws Exception {//删除临时有序节点zooKeeper.delete(this.lockPath,‐1);zooKeeper.close();System.out.println("锁已经释放:"+this.lockPath);}public static void main(String[] args) {try {MyLock myLock = new MyLock();myLock.createLock();} catch (Exception ex) {ex.printStackTrace();}}
}

public class TicketSeller {private void sell(){System.out.println("售票开始");// 线程随机休眠数毫秒，模拟现实中的费时操作int sleepMillis = 5000;try {//代表复杂逻辑执行了一段时间Thread.sleep(sleepMillis);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("售票结束");}public void sellTicketWithLock() throws Exception {MyLock lock = new MyLock();// 获取锁lock.acquireLock();sell();//释放锁lock.releaseLock();}public static void main(String[] args) throws Exception {TicketSeller ticketSeller = new TicketSeller();for(int i=0;i<10;i++){ticketSeller.sellTicketWithLock();}}
}

zooKeeper分布式锁原理

  • 核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除该节点。
  • 客户端获取锁时，在lock节点下创建临时顺序节点。
  • 然后获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。
  • 如果发现自己创建的节点并非lock所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。
  • 如果发现比自己小的那个节点被删除，则客户端的Watcher会收到相应通知，此时再次判断自己创建的节点是否是lock子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

二、zookeeper 集群搭建

2.1、搭建要求

真实的集群是需要部署在不同的服务器上的，但是在我们测试时同时启动很多个虚拟机内存会吃不消，所以我们通常会搭建伪集群，也就是把所有的服务都搭建在一台虚拟机上，用端口进行区分。

我们这里要求搭建一个三个节点的Zookeeper集群（伪集群）。

2.2、准备工作

重新部署一台虚拟机作为我们搭建集群的测试服务器。

（1）安装JDK 【此步骤省略】。

（2）Zookeeper压缩包上传到服务器
（3）将Zookeeper解压 ，建立/usr/local/zookeeper-cluster目录，将解压后的Zookeeper复制到以下三个目录

/usr/local/zookeeper-cluster/zookeeper-1

/usr/local/zookeeper-cluster/zookeeper-2

/usr/local/zookeeper-cluster/zookeeper-3

[root@localhost ~]# mkdir /usr/local/zookeeper-cluster
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-1
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-2
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-3

（4）创建data目录 ，并且将 conf下zoo_sample.cfg 文件改名为 zoo.cfg

mkdir /usr/local/zookeeper-cluster/zookeeper-1/data
mkdir /usr/local/zookeeper-cluster/zookeeper-2/data
mkdir /usr/local/zookeeper-cluster/zookeeper-3/datamv  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
mv  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
mv  /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

（5） 配置每一个Zookeeper 的dataDir 和 clientPort 分别为2181 2182 2183

修改/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfgclientPort=2181
dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data

修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfgclientPort=2182
dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data

修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfgclientPort=2183
dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data

2.3、配置集群

（1）在每个zookeeper的 data 目录下创建一个 myid 文件，内容分别是1、2、3 。这个文件就是记录每个服务器的ID

echo 1 >/usr/local/zookeeper-cluster/zookeeper-1/data/myid
echo 2 >/usr/local/zookeeper-cluster/zookeeper-2/data/myid
echo 3 >/usr/local/zookeeper-cluster/zookeeper-3/data/myid

（2）在每一个zookeeper 的 zoo.cfg配置客户端访问端口（clientPort）和集群服务器IP列表。

集群服务器IP列表如下

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfgserver.1=192.168.149.135:2881:3881
server.2=192.168.149.135:2882:3882
server.3=192.168.149.135:2883:3883

解释：server.服务器ID=服务器IP地址：服务器之间通信端口：服务器之间投票选举端口

2.4、启动集群

启动集群就是分别启动每个实例。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start

在这里插入图片描述

启动后我们查询一下每个实例的运行状态

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

先查询第一个服务

在这里插入图片描述

Mode为follower表示是跟随者（从）

再查询第二个服务Mod 为leader表示是领导者（主）

在这里插入图片描述

查询第三个为跟随者（从）
在这里插入图片描述

2.5、模拟集群异常

（1）首先我们先测试如果是从服务器挂掉，会怎么样

把3号服务器停掉，观察1号和2号，发现状态并没有变化

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh stop
/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

在这里插入图片描述

由此得出结论，3个节点的集群，从服务器挂掉，集群正常

（2）我们再把1号服务器（从服务器）也停掉，查看2号（主服务器）的状态，发现已经停止运行了。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh stop
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

在这里插入图片描述

由此得出结论，3个节点的集群，2个从服务器都挂掉，主服务器也无法运行。因为可运行的机器没有超过集群总数量的半数。

（3）我们再次把1号服务器启动起来，发现2号服务器又开始正常工作了。而且依然是领导者。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

在这里插入图片描述

（4）我们把3号服务器也启动起来，把2号服务器停掉,停掉后观察1号和3号的状态。

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh stop/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

在这里插入图片描述

发现新的leader产生了~
由此我们得出结论，当集群中的主服务器挂了，集群中的其他服务器会自动进行选举状态，然后产生新得leader

（5）我们再次测试，当我们把2号服务器重新启动起来启动后，会发生什么？2号服务器会再次成为新的领导吗？我们看结果

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

在这里插入图片描述
在这里插入图片描述

我们会发现，2号服务器启动后依然是跟随者（从服务器），3号服务器依然是领导者（主服务器），没有撼动3号服务器的领导地位。

由此我们得出结论，当领导者产生后，再次有新服务器加入集群，不会影响到现任领导者。

2.6、zookeeper 集群角色

在ZooKeeper集群服中务中有三个角色：

  • Leader 领导者 ：

    • 处理事务请求
    • 集群内部各服务器的调度者

  • Follower 跟随者 ：

    • 处理客户端非事务请求，转发事务请求给Leader服务器
    • 参与Leader选举投票

  • Observer 观察者：

    • 处理客户端非事务请求，转发事务请求给Leader服务器

在这里插入图片描述

三、一致性协议:zab协议

zab协议 的全称是 Zookeeper Atomic Broadcast （zookeeper原子广播）。zookeeper 是通过 zab协议来保证分布式事务的最终一致性

  1. leader从客户端收到一个写请求
  2. leader生成一个新的事务并为这个事务生成一个唯一的ZXID
  3. leader将这个事务提议(propose)发送给所有的follows节点
  4. follower节点将收到的事务请求加入到历史队列(history queue)中,并发送ack给
    leader
  5. 当leader收到大多数follower（半数以上节点）的ack消息，leader会发送commit请
    求
  6. 当follower收到commit请求时，从历史队列中将事务请求commit

四、zookeeper的leader选举

4.1?服务器状态

  • looking：寻找leader状态。当服务器处于该状态时，它会认为当前集群中没有
  • leader，因此需要进入leader选举状态。
  • leading： 领导者状态。表明当前服务器角色是leader。
  • following： 跟随者状态。表明当前服务器角色是follower。
  • observing：观察者状态。表明当前服务器角色是observer。

4.2、服务器启动时期的leader选举

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

    1. 每个server发出一个投票。由于是初始情况，server1和server2都会将自己作为leader服务器来进行投票，每次投票会包含所推举的服务器的myid和zxid，使用(myid, zxid)来表示，此时server1的投票为(1, 0)，server2的投票为(2, 0)，然后各自将这个投票发给集群中其他机器。
    1. 集群中的每台服务器接收来自集群中各个服务器的投票。
    1. 处理投票。针对每一个投票，服务器都需要将别人的投票和自己的投票进行pk，pk规则如下
    • 检查zxid。zxid比较大的服务器优先作为leader。
    • 如果zxid相同，那么就比较myid。myid较大的服务器作为leader服务器。
    • 对于Server1而言，它的投票是(1, 0)，接收Server2的投票为(2, 0)，首先会比较两者的zxid，均为0，再比较myid，此时server2的myid最大，于是更新自己的投票为(2, 0)，然后重新投票，对于server2而言，其无须更新自己的投票，只是再次向集群中所有机器发出上一次投票信息即可。
    1. 统计投票。每次投票后，服务器都会统计投票信息，判断是否已经有过半机器接受到相同的投票信息，对于server1、server2而言，都统计出集群中已经有两台机器接受了(2, 0)的投票信息，此时便认为已经选出了leader
    1. 改变服务器状态。一旦确定了leader，每个服务器就会更新自己的状态，如果是follower，那么就变更为following，如果是leader，就变更为leading。

4.3、服务器运行时期的Leader选举

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

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

  1. 变更状态。leader挂后，余下的服务器都会将自己的服务器状态变更为looking，然后开始进入leader选举过程。

  2. 每个server会发出一个投票。在运行期间，每个服务器上的zxid可能不同，此时假定server1的zxid为122，server3的zxid为122，在第一轮投票中，server1和server3都会投自己，产生投票(1, 122)，(3, 122)，然后各自将投票发送给集群中所有机器。

  3. 接收来自各个服务器的投票。与启动时过程相同

  4. 处理投票。与启动时过程相同，此时，server3将会成为leader。

  5. 统计投票。与启动时过程相同。

  6. 改变服务器的状态。与启动时过程相同。

五、observer角色及其配置

observer角色特点：

  • 不参与集群的leader选举
  • 不参与集群中写数据时的ack反馈
  • 为了使用observer角色，在任何想变成observer角色的配置文件中加入如下配
    置：
peerType=observer

并在所有server的配置文件中，配置成observer模式的server的那行配置追加:observer，例如：

server.3=192.168.60.130:2289:3389:observer

六、zookeeperAPI连接集群

ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)
  • connectionString ： zooKeeper集合主机。
  • sessionTimeout ： 会话超时（以毫秒为单位）。
  • watcher ： 实现“监视器”界面的对象。ZooKeeper集合通过监视器对象返回连接状
    态。
public class ZookeeperConnection {public static void main(String[] args) {try {// 计数器对象CountDownLatch countDownLatch=new CountDownLatch(1);// arg1:服务器的ip和端口// arg2:客户端与服务器之间的会话超时时间 以毫秒为单位的// arg3:监视器对象ZooKeeper zooKeeper=new ZooKeeper("192.168.60.130:2181,192.168.60.130:2182,192.168.60.130:2183",5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(event.getState()==Event.KeeperState.SyncConnected){System.out.println("连接创建成功!");countDownLatch.countDown();}}});// 主线程阻塞等待连接对象的创建成功countDownLatch.await();// 会话编号System.out.println(zooKeeper.getSessionId());zooKeeper.close();} catch (Exception ex) {ex.printStackTrace();}}
}

七、zookeeper四字监控命令

zooKeeper支持某些特定的四字命令与其的交互。它们大多是查询命令，用来获取 zooKeeper服务的当前状态及相关信息。用户在客户端可以通过 telnet 或 nc 向zooKeeper提交相应的命令。 zooKeeper常用四字命令见下表 所示：

命令描述
conf输出相关服务配置的详细信息。比如端口、zk数据及日志配置路径、最大连接数，session超时时间、serverId等
cons列出所有连接到这台服务器的客户端连接/会话的详细信息。包括“接受/发送”的包数量、session id 、操作延迟、最后的操作执行等信息
crst重置当前这台服务器所有连接/会话的统计信息
dump列出未经处理的会话和临时节点
envi输出关于服务器的环境详细信息
ruok测试服务是否处于正确运行状态。如果正常返回"imok"，否则返回空
stat输出服务器的详细信息：接收/发送包数量、连接数、模式（leader/follower）、节点总数、延迟。 所有客户端的列表
srst重置server状态
wchs列出服务器watches的简洁信息：连接总数、watching节点总数和watches总数
wchc通过session分组，列出watch的所有节点，它的输出是一个与 watch 相关的会话的节点列表
mntr列出集群的健康状态。包括“接受/发送”的包数量、操作延迟、当前服务模式（leader/follower）、节点总数、watch总数、临时节点总数

nc命令工具安装:

#root用户安装
#下载安装包
wget http://vault.centos.org/6.6/os/x86_64/Packages/nc-1.84-
22.el6.x86_64.rpm
#rpm安装
rpm -iUv nc-1.84-22.el6.x86_64.rpm

使用方式，在shell终端输入：echo mntr | nc localhost 2181

九、taokeeper监控工具的使用

基于zookeeper的监控管理工具taokeeper，由淘宝团队开源的zk管理中间件，安装前要求服务前先配置nc 和 sshd

1.下载数据库脚本

wget https://github.com/downloads/alibaba/taokeeper/taokeeper.sql

2.下载主程序

wget https://github.com/downloads/alibaba/taokeeper/taokeepermonitor.tar.gz

3.下载配置文件

wget https://github.com/downloads/alibaba/taokeeper/taokeeper-monitorconfig.propertie

4.配置 taokeeper-monitor-config.properties

#Daily
systemInfo.envName=DAILY
#DBCP
dbcp.driverClassName=com.mysql.jdbc.Driver
#mysql连接的ip地址端口号
dbcp.dbJDBCUrl=jdbc:mysql://192.168.60.130:3306/taokeeper
dbcp.characterEncoding=GBK
#用户名
dbcp.username=root
#密码
dbcp.password=root
dbcp.maxActive=30
dbcp.maxIdle=10
dbcp.maxWait=10000
#SystemConstant
#用户存储内部数据的文件夹
#创建/home/zookeeper/taokeeperdata/ZooKeeperClientThroughputStat
SystemConstent.dataStoreBasePath=/home/zookeeper/taokeeperdata
#ssh用户
SystemConstant.userNameOfSSH=zookeeper
#ssh密码
SystemConstant.passwordOfSSH=zookeeper
#Optional
SystemConstant.portOfSSH=22

5.安装配置 tomcat，修改catalina.sh

#指向配置文件所在的位置
JAVA_OPTS=-DconfigFilePath="/home/zookeeper/taokeeper-monitortomcat/webapps/ROOT/conf/taokeeper-monitor-config.properties

6.部署工程启动

在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/24087.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

epoll、poll、select的原理和区别

epoll、poll、select的原理和区别

select&#xff0c;poll&#xff0c;epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制&#xff0c;一个进程可以监视多个描述符&#xff0c;一旦某个描述符就绪&#xff08;一般是读就绪或者写就绪&#xff09;&#xff0c;能够通知程序进行相应的读写操作。但select&a…
阅读更多...
Markdown学习简记

Markdown学习简记

目录 一、写Markdown的第0步 二、语法须知 标题 粗体强调 斜体 斜体同时粗体 删除线 高亮 代码 代码块 引用块 无序列表 有序列表 链接 表格 图片 分割线 目录生成 内联HTML代码 Typora常用快捷键 Typora的主题样式与检查元素 一、写Markdown的第…
阅读更多...
Django Rest_Framework（三）

Django Rest_Framework（三）

文章目录 1. 认证Authentication2. 权限Permissions使用提供的权限举例自定义权限 3. 限流Throttling基本使用可选限流类 4. 过滤Filtering5. 排序Ordering6. 分页Pagination可选分页器 7. 异常处理 ExceptionsREST framework定义的异常 8. 自动生成接口文档coreapi安装依赖设置…
阅读更多...
Javaweb学习（2）

Javaweb学习（2）

Javaweb学习 一、Maven1.1 Maven概述1.2 Maven简介1.3、Maven基本使用1.4、IDEA配置Maven1.6、依赖管理&依赖范围 二、MyBatis2.1 MyBatis简介2.2 Mybatis快速入门2.3、解决SQL映射文件的警告提示2.4、Mapper代理开发 三、MyBaits核心配置文件四、 配置文件的增删改查4.1 M…
阅读更多...
Flink之RedisSink

Flink之RedisSink

在Flink开发中经常会有将数据写入到redis的需求,但是Flink官方并没有对应的扩展包,这个时候需要我们自己编译对应的jar资源,这个时候就用到了bahir,barhir是apahce的开源项目,是专门给spark和flink提供扩展包使用的,bahir官网,这篇文章就介绍下如何自己编译RedisSink扩展包. 下…
阅读更多...
在tensorflow分布式训练过程中突然终止（终止）

在tensorflow分布式训练过程中突然终止（终止）

问题 这是为那些将从服务器接收渐变的员工提供的培训功能&#xff0c;在计算权重和偏差后&#xff0c;将更新的渐变发送到服务器。代码如下&#xff1a; def train():"""Train CIFAR-10 for a number of steps."""g1 tf.Graph()with g1.as_de…
阅读更多...
【多线程初阶】多线程案例之单例模式

【多线程初阶】多线程案例之单例模式

文章目录 前言1. 什么是单例模式2. 饿汉模式3. 懒汉模式 --- 单线程版4. 懒汉模式 --- 多线程版5. 懒汉模式 --- 多线程改进版总结 前言 本文主要给大家讲解多线程的一个重要案例 — 单例模式. 关注收藏, 开始学习吧&#x1f9d0; 1. 什么是单例模式 单例模式是一种很经典的…
阅读更多...
简要介绍 | 生成模型的演进：从自编码器(AE)到变分自编码器(VAE)和生成对抗网络(GAN)，再到扩散模型

简要介绍 | 生成模型的演进：从自编码器(AE)到变分自编码器(VAE)和生成对抗网络(GAN)，再到扩散模型

注1：本文系“简要介绍”系列之一，仅从概念上对生成模型（包括AE, VAE, GAN，以及扩散模型）进行非常简要的介绍，不适合用于深入和详细的了解。 生成模型的演进：从自编码器(AE)到变分自编码器(VAE)和生成对抗网络(GAN)，再到扩散模型 一、背景介绍 生成模型在机器学习领域…
阅读更多...
blender 毛发粒子

blender 毛发粒子

新建平面&#xff0c;点击右侧粒子系统&#xff0c;选择毛发&#xff0c;调整毛发长度&#xff0c;数量&#xff08;Number&#xff09;&#xff0c;调整数量是为了避免电脑卡顿&#xff1b; 上面设置的每一根柱子都可以变成一个物体&#xff0c;点击渲染&#xff0c;渲染为选择…
阅读更多...
c++基本数据结构

c++基本数据结构

void insert(const node *head, node *p) {node *x, *y;yhead;do{xy;yx->next;} while ((y!NULL) && (y->value < p->value);x->nextp;p->nexty; } 二.栈 (1) 栈的实现! 操作规则&#xff1a;先进后出&#xff0c;先出后进。 int stack[N], top0; /…
阅读更多...
人工智能学习07--pytorch23--目标检测：Deformable-DETR训练自己的数据集

人工智能学习07--pytorch23--目标检测：Deformable-DETR训练自己的数据集

参考 https://blog.csdn.net/qq_44808827/article/details/125326909https://blog.csdn.net/dystsp/article/details/125949720?utm_mediumdistribute.pc_relevant.none-task-blog-2~default~baidujs_baidulandingword~default-0-125949720-blog-125326909.235^v38^pc_releva…
阅读更多...
JavaEE 面试常见问题

JavaEE 面试常见问题

一、常见的 ORM 框架有哪些&#xff1f; 1.Mybatis Mybatis 是一种典型的半自动的 ORM 框架&#xff0c;所谓的半自动&#xff0c;是因为还需要手动的写 SQL 语句&#xff0c;再由框架根据 SQL 及 传入数据来组装为要执行的 SQL 。其优点为&#xff1a; 1. 因为由程序员…
阅读更多...
【转】金融行业JR/T0197-2020《金融数据安全 数据安全分级指南》解读

【转】金融行业JR/T0197-2020《金融数据安全 数据安全分级指南》解读

原文链接&#xff1a;金融行业JR/T0197-2020《金融数据安全 数据安全分级指南》解读 《金融数据安全 数据安全分级指南》 解 读 随着IT技术的发展&#xff0c;银行的基础业务、核心流程等众多事务和活动都运营在信息化基础之上&#xff0c;金融机构运行过程中产生了大量的数字…
阅读更多...
词嵌入、情感分类任务

词嵌入、情感分类任务

目录 1.词嵌入&#xff08;word embedding&#xff09; 对单词使用one-hot编码的缺点是难以看出词与词之间的关系。 所以需要使用更加特征化的表示&#xff08;featurized representation&#xff09;&#xff0c;如下图所示&#xff0c;我们可以得到每个词的向量表达。 假设…
阅读更多...
IO（JavaEE初阶系列8）

IO（JavaEE初阶系列8）

目录 前言&#xff1a; 1.文件 1.1认识文件 1.2结构和目录 1.3文件路径 1.4文本文件vs二进制文件 2.文件系统的操作 2.1Java中操作文件 2.2File概述 2.2.1构造File对象 2.2.2File中的一些方法 3.文件内容的操作 3.1字节流 3.1.1InPutStream的使用方法 3.1.2OutPu…
阅读更多...
windows下安装anaconda、pycharm、cuda、cudnn、PyTorch-GPU版本

windows下安装anaconda、pycharm、cuda、cudnn、PyTorch-GPU版本

目录 一、anaconda安装及虚拟环境创建 1.anaconda的下载 2.Anaconda的安装 3.创建虚拟环境 3.1 环境启动 3.2 切换镜像源 3.3环境创建 3.4 激活环境 3.5删除环境 二、pycharm安装 1.pycharm下载 2.pycharm的安装 三、CUDA的安装 1.GPU版本和CUDA版本、cudnn版本、显卡…
阅读更多...
一起学算法（二维数组篇）

一起学算法（二维数组篇）

1.概念定义 1.矩阵的定义 矩阵A(nm)的定义时按照长方形排列的复数或实数集合&#xff0c;其中n代表的是行数&#xff0c;m代表的是列数。如下所示&#xff0c;代表的是一个4x3的矩阵 在Java中&#xff0c;我们可以用A[n][m]来代表一个n*m的矩阵&#xff0c;其中A[i][j]代表的是…
阅读更多...
python：基于Kalman滤波器的移动物体位置估计

python：基于Kalman滤波器的移动物体位置估计

CSDN@_养乐多_ Kalman滤波器是一种经典的估计方法，广泛应用于估计系统状态的问题。本篇博客将介绍Kalman滤波器的基本原理，并通过一个简单的Python代码示例，演示如何使用Kalman滤波器来估计移动物体的位置。 通过运行代码，我们将得到一个包含两个子图的图像，分别展示了估…
阅读更多...
第二十二篇：思路拓展：如何打造高性能的 React 应用？

第二十二篇：思路拓展：如何打造高性能的 React 应用？

React 应用也是前端应用&#xff0c;如果之前你知道一些前端项目普适的性能优化手段&#xff0c;比如资源加载过程中的优化、减少重绘与回流、服务端渲染、启用 CDN 等&#xff0c;那么这些手段对于 React 来说也是同样奏效的。 不过对于 React 项目来说&#xff0c;它有一个区…
阅读更多...
Ubuntu 23.04 作为系统盘的体验和使用感受

Ubuntu 23.04 作为系统盘的体验和使用感受

1.为啥主系统装了Ubuntu 由于公司发电脑了&#xff0c;我自己也有一台台式电脑&#xff0c;然后也想去折腾一下Ubuntu&#xff0c;就把自己的笔记本装成Ubuntu系统了&#xff0c; 我使用的是23.04的桌面版&#xff0c;带图形化界面的。我准备换回Windows 11了&#xff08;因为…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023