Zookeeper学习总结

IDEA环境

pom.xml

1<dependencies>
2	<dependency>
3		<groupId>org.apache.logging.log4j</groupId>
4		<artifactId>log4j-core</artifactId>
5		<version>2.8.2</version>
6	</dependency>
7
8	<dependency>
9		<groupId>org.apache.zookeeper</groupId>
10		<artifactId>zookeeper</artifactId>
11		<version>3.5.7</version>
12	</dependency>
13</dependencies>

log4j.properties

1log4j.rootLogger=INFO, stdout  
2log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
3log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
4log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
5log4j.appender.logfile=org.apache.log4j.FileAppender  
6log4j.appender.logfile.File=target/spring.log  
7log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
8log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n  

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

先在集群创建/servers节点

1create /servers "servers"

服务端向Zookeeper注册代码

1import org.apache.zookeeper.*;
2
3public class Server {
4    private String connectString;
5    private int sessionTimeOut;
6    private ZooKeeper zooKeeper;
7    private String parentNode = "/servers";
8
9    //创建客户端连接
10    public void getConnect() throws Exception{
11        connectString = "node01:2181,node02:22181,node03:2181";
12        sessionTimeOut = 10000;
13
14        zooKeeper = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
15            public void process(WatchedEvent watchedEvent) {
16
17            }
18        });
19    }
20
21    //注册服务器
22    public void registerServer(String hostname) throws Exception {
23        String create = zooKeeper.create(parentNode +"/server", hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
24
25        System.out.println(hostname + " is online " + create);
26    }
27
28    //业务功能
29    public void business(String hostname) throws Exception {
30        System.out.println(hostname + " is working...");
31        Thread.sleep(Long.MAX_VALUE);
32    }
33
34    public static void main(String[] args) throws Exception {
35        //获取连接
36        Server server = new Server();
37        server.getConnect();
38
39        //注册服务信息
40        server.registerServer(args[0]);
41
42        //启动业务功能
43        server.business(args[0]);
44    }
45}

客户端代码

1import org.apache.zookeeper.*;
2import java.util.ArrayList;
3import java.util.List;
4
5public class Client {
6    private String connectString;
7    private int sessionTimeOut;
8    private ZooKeeper zooKeeper;
9    private String parentNode = "/servers";
10
11    //创建客户端连接
12    public void getConnect() throws Exception{
13        connectString = "node01:2181,node02:22181,node03:2181";
14        sessionTimeOut = 10000;
15
16        zooKeeper = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
17            public void process(WatchedEvent watchedEvent) {
18                //再次启动监听
19                try {
20                    getServerList();
21                } catch (Exception e) {
22                    e.printStackTrace();
23                }
24            }
25        });
26    }
27
28    //获取服务器列表信息
29    public void getServerList() throws Exception {
30        //获取服务器子节点信息，并对父节点进行监听
31        List<String> children = zooKeeper.getChildren(parentNode, true);
32        //存储服务器信息列表
33        ArrayList<String> servers = new ArrayList<String>();
34        //遍历所有节点，获取节点中的主机名称
35        for (String child : children) {
36            byte[] data = zooKeeper.getData(parentNode + "/" + child, false, null);
37            servers.add(new String(data));
38        }
39        //打印服务器列表信息
40        System.out.println(servers);
41    }
42
43    //业务功能
44    public void business() throws Exception {
45        System.out.println("client is working...");
46        Thread.sleep(Long.MAX_VALUE);
47    }
48
49    public static void main(String[] args) throws Exception {
50        //获取zk连接
51        Client client = new Client();
52        client.getConnect();
53        //获取servers的子节点信息，从中获取服务器列表
54        client.getServerList();
55        //业务进程启动
56        client.business();
57    }
58}

测试

• 启动客户端

• 在node01上zk客户端的/servers目录上创建临时序列化节点,观察控制台变化

  1create -e -s /servers/node02 "node02"
  2create -e -s /servers/node03 "node03"
  3delete /servers/node020000000000
  

• 启动服务端

• 在IDEA运行配置中加入参数 node02

• 运行main方法，查看控制台

选举机制

节点信息说明

• SID：服务器ID，作为服务器的唯一标识，不能重复
• ZXID：事务ID，记录了修改的顺序，如果zxid1小于zxid2，那么zxid1在zxid2之前发生
• Epoch：每个Leader任期的代号，初始为0，每投一次票这个数就会增加

第一次启动

• 假设有5台机器，myid分别为1，2，...，5
• 服务器1启动，投自己一票，状态为LOOKING
• 服务器2启动，投自己一票，状态为LOOKING
• 服务器1与服务器2交换选票信息
• 由于服务器2的myid大于1的myid，1把票投给2
• 此时服务器1 0票，服务器2 2票
• 服务器3启动，按上述逻辑，服务器1，2 0票，服务器3 3票
• 由于服务器3票数过半，状态变为LEADER，1,2状态改为FLLOWING
• 服务器4启动，由于已经有LEADER了，所以服务器4状态变为FLLOWING
• 服务器5和服务器4一样

非第一次启动

• 如果存在LEADER，只需要和LEADER建立连接同步状态即可
• 如果集群中不存在LEADER，则按照以下规则进行选举
  • EPOCH大的直接胜出
  • EPOCH相同，ZXID大的胜出
  • ZXID也相同，SID（或myid，两个数值一样）大的胜出

监听器原理

• 首先要有一个main()线程
• main线程中创建Zookeeper客户端，这时会创建两个线程
• 一个负责通信(connect)，一个负责监听(listener)
• connect线程将注册的监听事件发送给Zookeeper
• 将注册的监听事件添加到Zookeeper的监听器列表中
• Zookeeper监听到数据或路径有变化，就会向listener线程发送消息
• listener内部重写process方法来响应变化

写数据流程

• 假设集群有三台机器，follower1,follower2,leader
• Client向follower1发送一个写的请求
• follower1会进一步把请求转发给leader
• leader收到请求后，将请求广播给各个follower
• 各个follower收到请求后会将请求加入写队列，并向leader发送成功信息
• 当leader收到半数以上的成功信息，说明写操作可以执行
• leader会向各个follower发送提交信息
• 各个follower收到消息后会将队列里的写请求执行，此时写成功

集群安装多少台zk合适

经验

• 10台服务器：3台zk

• 20台服务器：5台zk

• 100台服务器：11台zk

• 200台服务器：11台zk

为什么不装更多zk

• zk多，可以提高可靠性，但过多的zk会提高通信延迟

为什么要装奇数台

• 3台时，2台坏了就不能用了
• 4台时，2台坏了就不能用了
• 5台时，3台坏了就不能用了
• 6台时，3台坏了就不能用了
• ......
• 可以发现2n和2n-1的容忍度是一样的，没有必要多加一台zk