一篇搞定zookeeper

一、Zookeeper入门

1、概述

Zookeeper是一个开源的分布式的，为分布式应用提供协调服务的Apache项目。

Zookeeper从设计模式角度来理解，是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生了变化，Zookeeper就负责通知已经在Zookeeper上注册的那些观察者做出相应的反应.

Zookeeper = 文件系统 + 通知机制

2、特点

3、数据结构

4 、应用场景

提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。

统一命名服务：

统一配置管理：

统一集群管理：

服务器动态上下线：

负载均衡：

二、zookeeper安装

1、本地模式安装部署

1）安装前准备

（1）安装Jdk

（2）拷贝Zookeeper安装包到Linux系统下

（3）解压到指定目录

[atguigu@hadoop102 software]$ tar -zxvf zookeeper-3.5.7.tar.gz -C /opt/module/
1

2）配置修改

（1）将/opt/module/zookeeper-3.5.7/conf这个路径下的zoo_sample.cfg修改为zoo.cfg（这一步不是必须）；

[atguigu@hadoop102 conf]$ mv zoo_sample.cfg zoo.cfg
1

（2）打开zoo.cfg文件，修改dataDir路径：

[atguigu@hadoop102 zookeeper-3.5.7]$ vim zoo.cfg
1

修改如下内容：

dataDir=/opt/module/zookeeper-3.5.7/zkData
1

（3）在/opt/module/zookeeper-3.5.7/这个目录上创建zkData文件夹

[atguigu@hadoop102 zookeeper-3.5.7]$ mkdir zkData
1

3）操作Zookeeper

（1）启动Zookeeper

[atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkServer.sh start
1

（2）查看进程是否启动

[atguigu@hadoop102 zookeeper-3.5.7]$ jps
1

4020 Jps

4001 QuorumPeerMain

（3）查看状态：

[atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkServer.sh status
1

ZooKeeper JMX enabled by default

Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg

Mode: standalone
1
2
3
4
5

（4）启动客户端：

[atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkCli.sh
1

（5）退出客户端：

[zk: localhost:2181(CONNECTED) 0] quit
1

（6）停止Zookeeper

[atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkServer.sh stop
1

（可以配置PATH环境变量，以便于在其他目录下启停zookeeper）

2、 配置参数解读

Zookeeper中的配置文件zoo.cfg中参数含义解读如下：

1）tickTime =2000：通信心跳数，Zookeeper服务器与客户端心跳时间，单位毫秒

Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一个心跳，时间单位为毫秒。

它用于心跳机制，并且设置最小的session超时时间为两倍心跳时间。(session的最小超时时间是2*tickTime)

2）initLimit =10：LF初始通信时限

集群中的Follower跟随者服务器与Leader领导者服务器之间初始连接时能容忍的最多心跳数（tickTime的数量），用它来限定集群中的Zookeeper服务器连接到Leader的时限。

3）syncLimit =5：LF同步通信时限

集群中Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。

4）dataDir：数据文件目录+数据持久化路径

主要用于保存Zookeeper中的数据。

5）clientPort =2181：客户端连接端口

监听客户端连接的端口。

三、 Zookeeper实战（开发重点）

1、分布式安装部署

1）集群规划

在hadoop102、hadoop103和hadoop104三个节点上部署Zookeeper。

2）解压安装

（1）解压Zookeeper安装包到/opt/module/目录下

[atguigu@hadoop102 software]$ tar -zxvf zookeeper-3.5.7.tar.gz -C /opt/module/
1

（2）同步/opt/module/zookeeper-3.5.7目录内容到hadoop103、hadoop104

[atguigu@hadoop102 module]$ xsync zookeeper-3.5.7/
1

3）配置服务器编号

（1）在/opt/module/zookeeper-3.5.7/这个目录下创建zkData

[atguigu@hadoop102 zookeeper-3.5.7]$ mkdir -p zkData
1

（2）在/opt/module/zookeeper-3.5.7/zkData目录下创建一个myid的文件

[atguigu@hadoop102 zkData]$ touch myid
1

添加myid文件，注意一定要在linux里面创建，在notepad++里面很可能乱码

（3）编辑myid文件

[atguigu@hadoop102 zkData]$ vi myid
1

在文件中添加与server对应的编号：

2
1

（4）拷贝配置好的zookeeper到其他机器上

[atguigu@hadoop102 zkData]$ xsync myid
1

并分别在hadoop103、hadoop104上修改myid文件中内容为3、4

4）配置zoo.cfg文件

（1）重命名/opt/module/zookeeper-3.5.7/conf这个目录下的zoo_sample.cfg为zoo.cfg

[atguigu@hadoop102 conf]$ mv zoo_sample.cfg zoo.cfg
1

（2）打开zoo.cfg文件

[atguigu@hadoop102 conf]$ vim zoo.cfg
1

修改数据存储路径配置

dataDir=/opt/module/zookeeper-3.5.7/zkData
1

增加如下配置

#######################cluster##########################

server.2=hadoop102:2888:3888

server.3=hadoop103:2888:3888

server.4=hadoop104:2888:3888
1
2
3
4
5
6
7

（3）同步zoo.cfg配置文件

[atguigu@hadoop102 conf]$ xsync zoo.cfg
1

（4）配置参数解读

server.A=B:C:D。

A是一个数字，表示这个是第几号服务器；

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

B是这个服务器的地址；

C是这个服务器Follower与集群中的Leader服务器交换信息的端口；

D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

5）集群操作

（1）分别启动Zookeeper

[atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkServer.sh start

[atguigu@hadoop103 zookeeper-3.5.7]$ bin/zkServer.sh start

[atguigu@hadoop104 zookeeper-3.5.7]$ bin/zkServer.sh start
1
2
3
4
5

（2）查看状态

[atguigu@hadoop102 zookeeper-3.5.7]# bin/zkServer.sh status

JMX enabled by default

Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg

Mode: follower

[atguigu@hadoop103 zookeeper-3.5.7]# bin/zkServer.sh status

JMX enabled by default

Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg

Mode: leader

[atguigu@hadoop104 zookeeper-3.5.7]# bin/zkServer.sh status

JMX enabled by default

Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg

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

2、 客户端命令行操作

1）启动客户端

[atguigu@hadoop103 zookeeper-3.5.7]$ bin/zkCli.sh
（输入quit回车后退出客户端）
1
2

2）显示所有操作命令

[zk: localhost:2181(CONNECTED) 1] help
1

3）查看当前znode中所包含的内容

[zk: localhost:2181(CONNECTED) 0] ls /

[zookeeper]
1
2
3

4）查看当前节点详细数据

[zk: localhost:2181(CONNECTED) 1] ls -s /

[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  （0x0表示此节点不是临时节点，否则是临时节点）

dataLength = 0  （节点的数据长度）

numChildren = 1   （子节点的数量）
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

5）分别创建2个普通节点

[zk: localhost:2181(CONNECTED) 3] create /sanguo "diaochan"

Created /sanguo

[zk: localhost:2181(CONNECTED) 4] create /sanguo/shuguo "liubei"

Created /sanguo/shuguo
（ “” 中的值是节点的值，可以get读取，或者创建节点后set设置）
1
2
3
4
5
6
7
8

6）获得节点的值

[zk: localhost:2181(CONNECTED) 5] get /sanguo

diaochan

[zk: localhost:2181(CONNECTED) 6] get -s /sanguo

diaochan

cZxid = 0x100000003

ctime = Wed Aug 29 00:03:23 CST 2018

mZxid = 0x100000003

mtime = Wed Aug 29 00:03:23 CST 2018

pZxid = 0x100000004

cversion = 1

dataVersion = 0

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 7

numChildren = 1

[zk: localhost:2181(CONNECTED) 7]

[zk: localhost:2181(CONNECTED) 7] get -s /sanguo/shuguo

liubei

cZxid = 0x100000004

ctime = Wed Aug 29 00:04:35 CST 2018

mZxid = 0x100000004

mtime = Wed Aug 29 00:04:35 CST 2018

pZxid = 0x100000004

cversion = 0

dataVersion = 0

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 6

numChildren = 0
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

7）创建临时节点

[zk: localhost:2181(CONNECTED) 7] create -e /sanguo/wuguo "zhouyu"

Created /sanguo/wuguo
1
2
3

（1）在当前客户端是能查看到的

[zk: localhost:2181(CONNECTED) 3] ls /sanguo 

[wuguo, shuguo]
1
2
3

（2）退出当前客户端然后再重启客户端

[zk: localhost:2181(CONNECTED) 12] quit

[atguigu@hadoop104 zookeeper-3.5.7]$ bin/zkCli.sh
1
2
3

（3）再次查看根目录下短暂节点已经删除

[zk: localhost:2181(CONNECTED) 0] ls /sanguo

[shuguo]
1
2
3

8）创建带序号的节点

​ （1）先创建一个普通的根节点/sanguo/weiguo

[zk: localhost:2181(CONNECTED) 1] create /sanguo/weiguo "caocao"

Created /sanguo/weiguo
1
2
3

​ （2）创建带序号的节点

[zk: localhost:2181(CONNECTED) 2] create /sanguo/weiguo "caocao"

Node already exists: /sanguo/weiguo

[zk: localhost:2181(CONNECTED) 3] create -s /sanguo/weiguo "caocao"

Created /sanguo/weiguo0000000000

[zk: localhost:2181(CONNECTED) 4] create -s /sanguo/weiguo "caocao"

Created /sanguo/weiguo0000000001

[zk: localhost:2181(CONNECTED) 5] create -s /sanguo/weiguo "caocao"

Created /sanguo/weiguo0000000002

[zk: localhost:2181(CONNECTED) 6] ls /sanguo

[shuguo, weiguo, weiguo0000000000, weiguo0000000001, weiguo0000000002, wuguo]

[zk: localhost:2181(CONNECTED) 6]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

如果节点下原来没有子节点，序号从0开始依次递增。如果原节点下已有2个节点，则再排序时从2开始，以此类推。

9）修改节点数据值

[zk: localhost:2181(CONNECTED) 6] set /sanguo/weiguo "caopi"
1

10）节点的值变化监听（监听是一次性的，只会通知一次，再次改变不会通知）

​ （1）在hadoop104主机上注册监听/sanguo节点数据变化

 [zk: localhost:2181(CONNECTED) 8] get -w /sanguo
1

​ （2）在hadoop103主机上修改/sanguo节点的数据

[zk: localhost:2181(CONNECTED) 1] set /sanguo "xishi"
1

​ （3）观察hadoop104主机收到数据变化的监听

WATCHER::

WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo
1
2
3

11）节点的子节点变化监听（路径变化）

​ （1）在hadoop104主机上注册监听/sanguo节点的子节点变化

[zk: localhost:2181(CONNECTED) 1] ls -w /sanguo

[aa0000000001, server101]
1
2
3

​ （2）在hadoop103主机/sanguo节点上创建子节点

[zk: localhost:2181(CONNECTED) 2] create /sanguo/jin "simayi"

Created /sanguo/jin
1
2
3

​ （3）观察hadoop104主机收到子节点变化的监听

WATCHER::

WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo
1
2
3

12）删除节点（delete只能删除空节点，即节点下没有子节点）

[zk: localhost:2181(CONNECTED) 4] delete /sanguo/jin
1

13）递归删除节点

[zk: localhost:2181(CONNECTED) 15] deleteall /sanguo/shuguo
1

14）查看节点状态

[zk: localhost:2181(CONNECTED) 17] stat /sanguo

cZxid = 0x100000003

ctime = Wed Aug 29 00:03:23 CST 2018

mZxid = 0x100000011

mtime = Wed Aug 29 00:21:23 CST 2018

pZxid = 0x100000014

cversion = 9

dataVersion = 1

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 4

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

3、API应用

1）idea环境搭建

创建一个module，添加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.5.7</version>
		</dependency>
</dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

拷贝log4j.properties文件到项目根目录

需要在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入。

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender

log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

log4j.appender.logfile=org.apache.log4j.FileAppender

log4j.appender.logfile.File=target/spring.log

log4j.appender.logfile.layout=org.apache.log4j.PatternLayout

log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2）代码示例

package com.atguigu.zk;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

/**

- @author layne

- @create 2020-07-27 16:08

- 客户端代码常用套路

- 1.获取客户端对象

- 2.用客户端对象做各种操作

- 3.关闭客户端对象

- 最经典的案例: hdfs  zookeeper
  */
  public class ZkClinet {

  private String connectString;
  private int sessionTimeout;
  private ZooKeeper zkClient;

  @Before
  public void init() throws IOException {
      connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
      sessionTimeout = 4000;
      //1 获取客户端对象
      zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
          @Override
          public void process(WatchedEvent event) {


      }
  });


  }

  @After
  public void close() throws InterruptedException {
      //3 关闭客户端对象
      zkClient.close();
  }

  /**

  - 获取子节点列表,不监听
    */
    @Test
    public void ls() throws IOException, KeeperException, InterruptedException {

    //2.用客户端对象做各种操作
    List<String> children = zkClient.getChildren("/", false);
    System.out.println(children);

  }

  /**

  - 获取子节点列表,并监听
    */
    @Test
    public void lsAndWatch() throws KeeperException, InterruptedException {
    List<String> children = zkClient.getChildren("/", new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println(event);
            System.out.println("0523 is the best of atguigu");
        }
    });

    System.out.println(children);

    Thread.sleep(Long.MAX_VALUE);
    }

  /**

  - 创建子节点
    */
    @Test
    public void create() throws KeeperException, InterruptedException {
    //参数解读 1.创建节点的路径  2.节点的值  3.节点的权限(这个不用管,选个open完事) 4.节点的类型
    //zkClient.create("/atguigu","shangguigu".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
    zkClient.create("/atguigu2", "shangguigu".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

    Thread.sleep(10000);
    }

  /**

  - 获取子节点的值,不监听
    */
    @Test
    public void get() throws KeeperException, InterruptedException {
    //判断节点是否存在
    Stat stat = zkClient.exists("/atguigu", false);
    if (stat == null) {
        System.out.println("节点不存在...");
        return;
    }
    //能走到这,说明节点存在
    byte[] data = zkClient.getData("/atguigu", false, stat);

    System.out.println(new String(data));
    }

  /**

  - 获取节点的值,并且监听
    */
    @Test
    public void getAndWatch() throws KeeperException, InterruptedException {
    //判断节点是否存在
    Stat stat = zkClient.exists("/atguigu", false);
    if (stat == null) {
        System.out.println("节点不存在...");
        return;
    }

    byte[] data = zkClient.getData("/atguigu", new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println(event);
        }
    }, stat);
    System.out.println(new String(data));

    //线程阻塞,睡一会
    Thread.sleep(Long.MAX_VALUE);
    }

  /**

  - 设置节点的值
    */
    @Test
    public void set() throws KeeperException, InterruptedException {
    //判断节点是否存在
    Stat stat = zkClient.exists("/atguigu", false);
    if (stat == null) {
        System.out.println("节点不存在...");
        return;
    }
    zkClient.setData("/atguigu", "shang".getBytes(), stat.getVersion());
    }

  /**

  - 删除空节点
    */
    @Test
    public void delete() throws KeeperException, InterruptedException {
    //判断节点是否存在
    Stat stat = zkClient.exists("/aaa", false);
    if (stat == null) {
        System.out.println("节点不存在...");
        return;
    }
    zkClient.delete("/aaa", stat.getVersion());
    }

  /**

  - 删除非空节点,利用递归实现
    */
    public void deleteAll(String path, ZooKeeper zk) throws KeeperException, InterruptedException {
    //判断节点是否存在
    Stat stat = zkClient.exists(path, false);
    if (stat == null) {
        System.out.println("节点不存在...");
        return;
    }
    //如果传入的节点存在,获取一下传入节点的子节点
    List<String> children = zk.getChildren(path, false);
    //先判断,如果children为null,说明传入节点没有子节点
    if (children.isEmpty()) {
        zk.delete(path, stat.getVersion());
    } else {
        //如果传入节点有子节点,遍历
        for (String child : children) {
            //删除传入节点的子节点,因为不知道子节点下面还有没有孙子节点,所以要递归调用自己
            deleteAll(path + "/" + child,zk);   //path = /aaa   child=bbb   /aaa/bbb
        }
        //删除掉所有字几点以后,还要删除传入的节点
        zk.delete(path,stat.getVersion());
    }
    }

//写一个方法,测试一下deleteAll
@Test
public void testDeleteAll() throws KeeperException, InterruptedException {
    deleteAll("/aaa",zkClient);
}

}
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207

四、Zookeeper内部原理

1、节点类型

创建临时节点的客户端退出后，临时节点消失

2、Stat结构体

（1）czxid-创建节点的事务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。（0x0是持久节点，否则是临时节点）

（10）dataLength- znode的数据长度

（11）numChildren - znode子节点数量

3 、监听器原理（面试重点）

4、选举机制（面试重点）

（1）半数机制：集群中半数以上机器存活，集群可用。所以Zookeeper适合安装奇数台服务器。

（2）Zookeeper虽然在配置文件中并没有指定Leader和Follwer。但是，Zookeeper工作时，是有一个节点为Leader，其他则为Follower，Leader是通过内部的选举机制临时产生的。

（3）以一个简单的例子来说明整个选举的过程。

假设有五台服务器组成的Zookeeper集群，它们的id从1-5，同时它们都是最新启动的，也就是没有历史数据，在存放数据量这一点上，都是一样的。假设这些服务器依序启动，来看看会发生什么。

​ Zookeeper的选举机制

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

（2）服务器2启动，再发起一次选举。服务器1和2分别投自己一票并交换选票信息：此时服务器1发现服务器2的ID比自己目前投票推举的（服务器1）大，更改选票为推举服务器2。此时服务器1票数0票，服务器2票数2票，没有半数以上结果，选举无法完成，服务器1，2状态保持LOOKING

（3）服务器3启动，发起一次选举。此时服务器1和2都会更改选票为服务器3。此次投票结果：服务器1为0票，服务器2为0票，服务器3为3票。此时服务器3的票数已经超过半数，服务器3当选Leader。服务器1，2更改状态为FOLLOWING，服务器3更改状态为LEADING；

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

（5）服务器5启动，同4一样当小弟。

5、写数据流程

（leader自己也会写）

五、企业面试真题

1、 请简述ZooKeeper的选举机制

详见4.4。

2、ZooKeeper的监听原理是什么？

详见4.3。

3、ZooKeeper的部署方式有哪几种？集群中的角色有哪些？集群最少需要几台机器？

（1）部署方式单机模式、集群模式

（2）角色：Leader和Follower

（3）集群最少需要机器数：3

4、ZooKeeper的常用命令

ls 、create 、get 、delete 、set等等