第3章 Zookeeper内部原理 API实战

太过爱你忘了你带给我的痛 2022-12-29 06:47 123阅读 0赞

### 大数据技术之Zookeeper ###

*  第3章 Zookeeper内部原理
 *   *  3.1 选举机制
     *  3.2 节点类型
     *   *  1．Znode有两种类型
         *  2．Znode有四种形式的目录节点（默认是persistent ）
     *  3.3 stat结构体
     *  3.4 监听器原理
     *  3.5 写数据流程
 *  第4章 Zookeeper实战
 *   *  4.1 分布式安装部署
     *  4.2 客户端命令行操作
 *  4.3 API应用
 *   *   *  监听服务器节点动态上下线案例

# 第3章 Zookeeper内部原理 #

## 3.1 选举机制 ##

> 1）半数机制（Paxos 协议）：集群中半数以上机器存活，集群可用。所以zookeeper适合装在奇数台机器上。  
> 2）Zookeeper虽然在配置文件中并没有指定master和slave但是，zookeeper工作时，是有一个节点为leader，其他则为follower，Leader是通过内部的选举机制临时产生的  
> 3）以一个简单的例子来说明整个选举的过程。  
> 假设有五台服务器组成的zookeeper集群，它们的id从1-5，同时它们都是最新启动的，也就是没有历史数据，在存放数据量这一点上，都是一样的。假设这些服务器依序启动，来看看会发生什么，如图5-8所示。

![图5-8 Zookeeper的选举机制][5-8 Zookeeper]

> （1）服务器1启动，此时只有它一台服务器启动了，它发出去的报没有任何响应，所以它的选举状态一直是LOOKING状态。  
> （2）服务器2启动，它与最开始启动的服务器1进行通信，互相交换自己的选举结果，由于两者都没有历史数据，所以id值较大的服务器2胜出，但是由于没有达到超过半数以上的服务器都同意选举它(这个例子中的半数以上是3)，所以服务器1、2还是继续保持LOOKING状态。  
> （3）服务器3启动，根据前面的理论分析，服务器3成为服务器1、2、3中的老大，而与上面不同的是，此时有三台服务器选举了它，所以它成为了这次选举的leader。  
> （4）服务器4启动，根据前面的分析，理论上服务器4应该是服务器1、2、3、4中最大的，但是由于前面已经有半数以上的服务器选举了服务器3，所以它只能接收当小弟的命了。  
> （5）服务器5启动，同4一样当小弟。

## 3.2 节点类型 ##

### 1．Znode有两种类型 ###

短暂（ephemeral） 客户端和服务器端断开连接后，创建的节点自己删除  
持久（persistent） 客户端和服务器端断开连接后，创建的节点不删除

### 2．Znode有四种形式的目录节点（默认是persistent ） ###

（1）持久化目录节点（PERSISTENT）  
客户端与zookeeper断开连接后，该节点依旧存在  
（2）持久化顺序编号目录节点（PERSISTENT\_SEQUENTIAL）  
客户端与zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号  
（3）临时目录节点（EPHEMERAL）  
客户端与zookeeper断开连接后，该节点被删除  
（4）临时顺序编号目录节点（EPHEMERAL\_SEQUENTIAL）  
客户端与zookeeper断开连接后，该节点被删除，只是Zookeeper给该节点名称进

行顺序编号，如图5-9所示  
![图5-9 Zookeeper节点类型][5-9 Zookeeper]  
***3．创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护  
4．在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序***

## 3.3 stat结构体 ##

> 1）czxid- 引起这个znode创建的zxid，创建节点的事务的zxid  
> 每次修改ZooKeeper状态都会收到一个zxid形式的时间戳，也就是ZooKeeper事务ID。  
> 事务ID是ZooKeeper中所有修改总的次序。每个修改都有唯一的zxid，如果zxid1小于zxid2，那么zxid1在zxid2之前发生。  
> 2）ctime - znode被创建的毫秒数(从1970年开始)  
> 3）mzxid - znode最后更新的zxid  
> 4）mtime - znode最后修改的毫秒数(从1970年开始)  
> 5）pZxid-znode最后更新的子节点zxid  
> 6）cversion - znode子节点变化号，znode子节点修改次数  
> 7）dataversion - znode数据变化号  
> 8）aclVersion - znode访问控制列表的变化号  
> 9）ephemeralOwner- 如果是临时节点，这个是znode拥有者的session id。如果不是临时节点则是0。  
> 10）dataLength- znode的数据长度  
> 11）numChildren - znode子节点数量

## 3.4 监听器原理 ##

![图5-10 监听器原理][5-10]

> 1．监听原理详解，如图5-10所示  
> 1）首先要有一个main()线程  
> 2）在main线程中创建Zookeeper客户端，这时就会创建两个线程，一个负责网络连接通信（connet），一个负责监听（listener）。  
> 3）通过connect线程将注册的监听事件发送给Zookeeper。  
> 4）在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中。  
> 5）Zookeeper监听到有数据或路径变化，就会将这个消息发送给listener线程。  
> 6）listener线程内部调用了process（）方法。

> 2．常见的监听  
> （1）监听节点数据的变化：  
> get path \[watch\]  
> （2）监听子节点增减的变化  
> ls path \[watch\]

## 3.5 写数据流程 ##

![图5-11 Zookeeper写数据流程][5-11 Zookeeper]  
ZooKeeper 的写数据流程主要分为以下几步，如图5-11所示：

1）比如 Client 向 ZooKeeper 的 Server1 上写数据，发送一个写请求。  
2）如果Server1不是Leader，那么Server1 会把接受到的请求进一步转发给Leader，因为每个ZooKeeper的Server里面有一个是Leader。这个Leader 会将写请求广播给各个Server，比如Server1和Server2， 各个Server写成功后就会通知Leader。  
3）当Leader收到大多数 Server 数据写成功了，那么就说明数据写成功了。如果这里三个节点的话，只要有两个节点数据写成功了，那么就认为数据写成功了。写成功之后，Leader会告诉Server1数据写成功了。  
4）Server1会进一步通知 Client 数据写成功了，这时就认为整个写操作成功。ZooKeeper 整个写数据流程就是这样的。

# 第4章 Zookeeper实战 #

## 4.1 分布式安装部署 ##

1．集群规划  
**在master、server01和server02三个节点上部署Zookeeper。**  
2．解压安装  
1）解压zookeeper安装包到/home/zhiquan/modules/目录下

> \[atguigu@master software\]$ tar -zxvf zookeeper-3.4.10.tar.gz -C /home/zhiquan/modules/

> （2）在/home/zhiquan/modules/zookeeper-3.4.10/这个目录下创建zkData  
> mkdir -p zkData

> （3）重命名/home/zhiquan/modules/zookeeper-3.4.10/conf这个目录下的zoo\_sample.cfg为zoo.cfg  
> mv zoo\_sample.cfg zoo.cfg

3．配置zoo.cfg文件

> （1）具体配置  
> dataDir=/opt/module/zookeeper-3.4.10/zkData  
> 增加如下配置  
> \#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#cluster\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#  
> server.2=hadoop102:2888:3888  
> server.3=hadoop103:2888:3888  
> server.4=hadoop104:2888:3888  
> （2）配置参数解读  
> server.A=B:C:D。

A是一个数字，表示这个是第几号服务器；  
B是这个服务器的ip地址或者主机名；  
C是这个服务器与集群中的Leader服务器交换信息的端口；  
D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

集群模式下配置一个文件myid，这个文件在dataDir目录下，这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。  
4．集群操作

> （1）在/home/zhiquan/modules/zookeeper-3.4.10/zkData目录下创建一个myid的文件  
> touch myid  
> 添加myid文件，注意一定要在linux里面创建，在notepad++里面很可能乱码  
> （2）编辑myid文件  
> vi myid  
> 在文件中添加与server对应的编号：如2  
> （3）拷贝配置好的zookeeper到其他机器上  
> scp -r zookeeper-3.4.10/ root@server01.atguigu.com:/home/zhiquan/app/  
> scp -r zookeeper-3.4.10/ root@server02.atguigu.com:/home/zhiquan/app/

并分别修改myid文件中内容为3、4

> （4）分别启动zookeeper  
> \[root@master zookeeper-3.4.10\]\# bin/zkServer.sh start  
> \[root@server01 zookeeper-3.4.10\]\# bin/zkServer.sh start  
> \[root@server02 zookeeper-3.4.10\]\# bin/zkServer.sh start  
> （5）查看状态  
> \[root@master zookeeper-3.4.10\]\# bin/zkServer.sh status  
> JMX enabled by default  
> Using config: /home/zhiquan/modules/zookeeper-3.4.10/bin/…/conf/zoo.cfg  
> Mode: follower  
> \[root@server01 zookeeper-3.4.10\]\# bin/zkServer.sh status  
> JMX enabled by default  
> Using config:/home/zhiquan/modules/zookeeper-3.4.10/bin/…/conf/zoo.cfg  
> Mode: leader  
> \[root@server02 zookeeper-3.4.5\]\# bin/zkServer.sh status  
> JMX enabled by default  
> Using config: /home/zhiquan/modules/zookeeper-3.4.10/bin/…/conf/zoo.cfg  
> Mode: follower

## 4.2 客户端命令行操作 ##

<table> 
 <thead> 
  <tr> 
   <th>命令基本语法</th> 
   <th>功能描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>help</td> 
   <td>显示所有操作命令</td> 
  </tr> 
  <tr> 
   <td>ls path [watch]</td> 
   <td>使用 ls 命令来查看当前znode中所包含的内容</td> 
  </tr> 
  <tr> 
   <td>ls2 path [watch]</td> 
   <td>查看当前节点数据并能看到更新次数等数据</td> 
  </tr> 
  <tr> 
   <td>create</td> 
   <td>普通创建</td> 
  </tr> 
  <tr> 
   <td>::</td> 
   <td>-s 含有序列</td> 
  </tr> 
  <tr> 
   <td>::</td> 
   <td>-e 临时（重启或者超时消失）</td> 
  </tr> 
  <tr> 
   <td>get path [watch]</td> 
   <td>获得节点的值</td> 
  </tr> 
  <tr> 
   <td>set</td> 
   <td>设置节点的具体值</td> 
  </tr> 
  <tr> 
   <td>stat</td> 
   <td>查看节点状态</td> 
  </tr> 
  <tr> 
   <td>delete</td> 
   <td>删除节点</td> 
  </tr> 
  <tr> 
   <td>rmr</td> 
   <td>递归删除节点</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

> 1．启动客户端  
> \[atguigu@hadoop103 zookeeper-3.4.10\]$ bin/zkCli.sh  
> 2．显示所有操作命令  
> \[zk: localhost:2181(CONNECTED) 1\] help  
> 3．查看当前znode中所包含的内容  
> \[zk: localhost:2181(CONNECTED) 0\] ls /  
> \[zookeeper\]  
> 4．查看当前节点数据并能看到更新次数等数据  
> \[zk: localhost:2181(CONNECTED) 1\] ls2 /  
> \[zookeeper\]  
> cZxid = 0x0  
> ctime = Thu Jan 01 08:00:00 CST 1970  
> mZxid = 0x0  
> mtime = Thu Jan 01 08:00:00 CST 1970  
> pZxid = 0x0  
> cversion = -1  
> dataVersion = 0  
> aclVersion = 0  
> ephemeralOwner = 0x0  
> dataLength = 0  
> numChildren = 1

> 5．创建普通节点  
> \[zk: localhost:2181(CONNECTED) 2\] create /app1 “hello app1”  
> Created /app1  
> \[zk: localhost:2181(CONNECTED) 4\] create /app1/server101 “192.168.1.101”  
> Created /app1/server101

> 6．获得节点的值  
> \[zk: localhost:2181(CONNECTED) 6\] get /app1  
> hello app1  
> cZxid = 0x20000000a  
> ctime = Mon Jul 17 16:08:35 CST 2017  
> mZxid = 0x20000000a  
> mtime = Mon Jul 17 16:08:35 CST 2017  
> pZxid = 0x20000000b  
> cversion = 1  
> dataVersion = 0  
> aclVersion = 0  
> ephemeralOwner = 0x0  
> dataLength = 10  
> numChildren = 1  
> \[zk: localhost:2181(CONNECTED) 8\] get /app1/server101  
> 192.168.1.101  
> cZxid = 0x20000000b  
> ctime = Mon Jul 17 16:11:04 CST 2017  
> mZxid = 0x20000000b  
> mtime = Mon Jul 17 16:11:04 CST 2017  
> pZxid = 0x20000000b  
> cversion = 0  
> dataVersion = 0  
> aclVersion = 0  
> ephemeralOwner = 0x0  
> dataLength = 13  
> numChildren = 0

> 7．创建短暂节点  
> \[zk: localhost:2181(CONNECTED) 9\] create -e /app-emphemeral 8888  
> （1）在当前客户端是能查看到的  
> \[zk: localhost:2181(CONNECTED) 10\] ls /  
> \[app1, app-emphemeral, zookeeper\]  
> （2）退出当前客户端然后再重启客户端  
> \[zk: localhost:2181(CONNECTED) 12\] quit  
> \[atguigu@server02 zookeeper-3.4.10\]$ bin/zkCli.sh  
> （3）再次查看根目录下短暂节点已经删除  
> \[zk: localhost:2181(CONNECTED) 0\] ls /  
> \[app1, zookeeper\]

> 8．创建带序号的节点  
> （1）先创建一个普通的根节点app2  
> \[zk: localhost:2181(CONNECTED) 11\] create /app2 “app2”  
> （2）创建带序号的节点  
> \[zk: localhost:2181(CONNECTED) 13\] create -s /app2/aa 888  
> Created /app2/aa0000000000  
> \[zk: localhost:2181(CONNECTED) 14\] create -s /app2/bb 888  
> Created /app2/bb0000000001  
> \[zk: localhost:2181(CONNECTED) 15\] create -s /app2/cc 888  
> Created /app2/cc0000000002  
> 如果原节点下有1个节点，则再排序时从1开始，以此类推。  
> \[zk: localhost:2181(CONNECTED) 16\] create -s /app1/aa 888  
> Created /app1/aa0000000001  
> 9．修改节点数据值  
> \[zk: localhost:2181(CONNECTED) 2\] set /app1 999  
> 10．节点的值变化监听  
> （1）在104主机上注册监听/app1节点数据变化  
> \[zk: localhost:2181(CONNECTED) 26\] get /app1 watch  
> （2）在103主机上修改/app1节点的数据  
> \[zk: localhost:2181(CONNECTED) 5\] set /app1 777  
> （3）观察104主机收到数据变化的监听  
> WATCHER::  
> WatchedEvent state:SyncConnected type:NodeDataChanged path:/app1  
> 11．节点的子节点变化监听（路径变化）  
> （1）在104主机上注册监听/app1节点的子节点变化  
> \[zk: localhost:2181(CONNECTED) 1\] ls /app1 watch  
> \[aa0000000001, server101\]  
> （2）在103主机/app1节点上创建子节点  
> \[zk: localhost:2181(CONNECTED) 6\] create /app1/bb 666  
> Created /app1/bb  
> （3）观察104主机收到子节点变化的监听  
> WATCHER::  
> WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/app1  
> 12．删除节点  
> \[zk: localhost:2181(CONNECTED) 4\] delete /app1/bb  
> 13．递归删除节点  
> \[zk: localhost:2181(CONNECTED) 7\] rmr /app2

> 14．查看节点状态  
> \[zk: localhost:2181(CONNECTED) 12\] stat /app1  
> cZxid = 0x20000000a  
> ctime = Mon Jul 17 16:08:35 CST 2017  
> mZxid = 0x200000018  
> mtime = Mon Jul 17 16:54:38 CST 2017  
> pZxid = 0x20000001c  
> cversion = 4  
> dataVersion = 2  
> aclVersion = 0  
> ephemeralOwner = 0x0  
> dataLength = 3  
> numChildren = 2

# 4.3 API应用 #

4.3.1 IDEA 环境搭建  
1．创建一个Maven工程  
2．添加pom文件 导入依赖

<dependencies>
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<version>RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.logging.log4j</groupId>
    			<artifactId>log4j-core</artifactId>
    			<version>2.8.2</version>
    		</dependency>
    		
    		<dependency>
    			<groupId>org.apache.zookeeper</groupId>
    			<artifactId>zookeeper</artifactId>
    			<version>3.4.10</version>
    		</dependency>
    	</dependencies>

3．拷贝log4j.properties文件到项目根目录  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70]  
4.3.2 创建ZooKeeper客户端

Public class  ZKDemo{
        
      private static String connectString =
        "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    	   private static int sessionTimeout = 2000;
         private ZooKeeper zkClient = null;
    
    	@Before
    	public void init() throws Exception {
        
    
    	zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        
    			@Override
    			public void process(WatchedEvent event) {
        
    				// 收到事件通知后的回调函数（用户的业务逻辑）
    				System.out.println(event.getType() + "--" + event.getPath());
    
    				// 再次启动监听
    				try {
        
    					zkClient.getChildren("/", true);
    				} catch (Exception e) {
        
    					e.printStackTrace();
    				}
    			}
    		});
    	}
    }

4.3.3 创建子节点

// 创建子节点
    	@Test
    	public void create() throws Exception {
        
    		// 数据的增删改查
    		// 参数1：要创建的节点的路径； 参数2：节点数据 ； 参数3：节点权限 ；参数4：节点的类型
    		String nodeCreated = zkClient.create("/eclipse", "hello zk".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
    	}

4.3.4 获取子节点并监听 (改造为循环监听)

// 获取子节点
    	@Test
    	public void getChildren() throws Exception {
        
    		List<String> children = zkClient.getChildren("/", true);
    
    		for (String child : children) {
        
    			System.out.println(child);
    		}
    
    		// 延时阻塞
    		Thread.sleep(Long.MAX_VALUE);
    	}

4.3.5 判断znode是否存在

// 判断znode是否存在
    	@Test
    	public void exist() throws Exception {
        
    		Stat stat = zkClient.exists("/eclipse", false);
    
    		System.out.println(stat == null ? "not exist" : "exist");
    	}

4.3.6 改变节点的内容

// 改变节点的内容
    	@Test
    	public void testSet() {
        
    		zooKeeper.setData(“/atguigu/0308”,”i love 0308”.getBytes(),-1)
    
    		System.out.println(“data changed”);
    	}

4.3.7 获取节点数据

// 获取节点数据
    	@Test
    	public void testGet() {
        
    		Byte[] data = zooKeeper.getData(“/atguigu/0308”,true.null)
    
    		System.out.println(new String(data));
    
            Thread.sleep(Long.)
    	}

### 监听服务器节点动态上下线案例 ###

1．需求  
某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线  
2．需求分析，如图5-12所示  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 1]  
3．具体实现  
（0）现在集群上创建/servers节点

> \[zk: localhost:2181(CONNECTED) 10\] create /servers “servers”  
> Created /servers

（1）服务器端代码

package com.atguigu.zkcase;
    import java.io.IOException;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.ZooDefs.Ids;
    
    public class DistributeServer {
        
    
    	private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    	private static int sessionTimeout = 2000;
    	private ZooKeeper zk = null;
    	private String parentNode = "/servers";
    	
    	// 创建到zk的客户端连接
    	public void getConnect() throws IOException{
        
    		
    		zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        
    
    			@Override
    			public void process(WatchedEvent event) {
        
                         Sout(event.setType()+event.getPath());
    			}
    		});
    	}
    	
    
    	// 注册服务器
    	public void registServer(String hostname) throws Exception{
        
    		String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    		
    		System.out.println(hostname +" is online "+ create);
    	}
    	
    	// 业务功能
    	public void business(String hostname) throws Exception{
        
    		System.out.println(hostname+" is working ...");
    		
    		Thread.sleep(Long.MAX_VALUE);
    	}
    	
    	public static void main(String[] args) throws Exception {
        
    		// 1获取zk连接
    		DistributeServer server = new DistributeServer();
    		server.getConnect();
    		
    		// 2 利用zk连接注册服务器信息
    		server.registServer(args[0]);
    		
    		// 3 启动业务功能
    		server.business(args[0]);
    	}
    }

（2）客户端代码

package com.atguigu.zkcase;
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    
    public class DistributeClient {
        
    	private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    	private static int sessionTimeout = 2000;
    	private ZooKeeper zk = null;
    	private String parentNode = "/servers";
    
    	// 创建到zk的客户端连接
    	public void getConnect() throws IOException {
        
    		zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        
    
    			@Override
    			public void process(WatchedEvent event) {
        
                       Sout(event.setType()+event.getPath());
    				// 再次启动监听
    				try {
        
    					getServerList();
    				} catch (Exception e) {
        
    					e.printStackTrace();
    				}
    			}
    		});
    	}
    
    	// 获取服务器列表信息
    	public void getServerList() throws Exception {
        
    		
    		// 1获取服务器子节点信息，并且对父节点进行监听
    		List<String> children = zk.getChildren(parentNode, true);
    
            // 2存储服务器信息列表
    		ArrayList<String> servers = new ArrayList<>();
    		
            // 3遍历所有节点，获取节点中的主机名称信息
    		for (String child : children) {
        
    			byte[] data = zk.getData(parentNode + "/" + child, false, null);
    
    			servers.add(new String(data));
    		}
    
            // 4打印服务器列表信息
    		System.out.println(servers);
    	}
    
    	// 业务功能
    	public void business() throws Exception {
        
    		System.out.println("client is working ...");
    Thread.sleep(Long.MAX_VALUE);
    	}
    
    	public static void main(String[] args) throws Exception {
        
    
    		// 1获取zk连接
    		DistributeClient client = new DistributeClient();
    		client.getConnect();
    
    		// 2获取servers的子节点信息，从中获取服务器信息列表
    		client.getServerList();
    
    		// 3业务进程启动
    		client.business();
    	}
    }

[5-8 Zookeeper]: /images/20221120/55d85b99931646c3bdd5e2aff9aab1da.png
[5-9 Zookeeper]: /images/20221120/aeed06e786e64bbdace8a29487404a9c.png
[5-10]: https://img-blog.csdnimg.cn/20201221092308362.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[5-11 Zookeeper]: https://img-blog.csdnimg.cn/20201221092445653.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20201221094335566.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20201221094623772.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6cTE4ODQ1MjgwNjEw,size_16,color_FFFFFF,t_70