昨天晚上看了一篇博客，作者实现了一个分布式的调度框架，其中支持两种集群模式，其中一种就是主备模式，是基于 ZooKeeper 实现的，这也是 ZooKeeper 很常见的应用场景，还没来得及看具体细节就去处理了一个线上问题，今天一直找不到那个博客链接。今天就尝试自己实现一下，本文会介绍两种实现方式（总体思想一致，部分细节有所差别）。

这种主备模式首先需要从集群中选出 Master 节点，然后剩余节点 Standby，当 Master 节点挂掉之后，剩余节点争抢或者根据一定的策略指定出 Master 节点，还有一个特点是这种模式干活的只有 Master 节点。

本文使用 Curator 操作 ZooKeeper，引入相关依赖：

<dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.0</version>
        </dependency>

环境准备：

好久没用 ZooKeeper 了，之前环境配置的都忘记了，还好不算折腾，这里简单记录下：

启动 ZooKeeper：

[root@localhost bin]# pwd
/usr/local/develope/zookeeper/zookeeper/bin
[root@localhost bin]# sh zkServer.sh start 

客户端连接：

[root@localhost bin]# sh zkCli.sh -server 127.0.0.1:2181

发现一直不停输出：

Unable to read additional data from server sessionid 0x0

这是因为我之前配置的是三台 ZooKeeper 集群，我这里只启动了一台机器，最少要启动半数以上的机器，所以要改一下 zoo.cfg 文件：

[root@localhost bin]# cd ..
[root@localhost bin]# cd conf
[root@localhost bin]# vim zoo.cfg 

这里只配置一个即可：

重新启动 ZooKeeper 服务，再进入客户端：

[root@localhost bin]# sh zkServer.sh stop 
[root@localhost bin]# sh zkServer.sh start
[root@localhost bin]# sh zkCli.sh -server 127.0.0.1:2181
[zk: 127.0.0.1:2181(CONNECTED) 0] ls
[zk: 127.0.0.1:2181(CONNECTED) 1] ls /
[zookeeper]

配置类：

@Configuration
public class ZooKeeperConfiguration {

    private static final String ZOOKEEPER_URL = "192.168.43.6:2181";

    @Bean
    public CuratorFramework getCuratorFramework(){
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(ZOOKEEPER_URL,retryPolicy);
        client.start();
        return client;
    }
}

第一种实现

这种实现方式就是基于 ZooKeeper 的临时节点，集群每个节点启动的时候都会去创建一个临时节点，创建成功了的节点被选为 Master，然后其余节点监控这个临时节点即可。

抢占 Master：

package com.example.simplespringboot.configuration;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @author Dongguabai
 * @Description
 * @Date 创建于 2020-05-05 18:54
 */
@Component
public class MasterRegister implements CommandLineRunner {

    private static final String ROOT_PATH = "/test";

    private static final Long WAIT_SECONDS = 3L;

    public volatile boolean master = false;

    @Autowired
    private CuratorFramework zkClient;

    @Value("${server.port}")
    private String port;

    @Override
    public void run(String... args) throws Exception {
        //Spring 容器启动后创建临时节点
        //由于在ZooKeeper中规定了所有非叶子节点必须为持久节点，调用上面这个API之后，只有path参数对应的数据节点是临时节点，其父节点均为持久节点
        regist();
        PathChildrenCache childrenCache = new PathChildrenCache(zkClient, ROOT_PATH , true);
        childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        // 节点数据change事件的通知方法
        childrenCache.getListenable().addListener((curatorFramework, event) -> {
            if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                System.out.println("节点变更，开始重新选举");
                //todo：防止脑裂/防止网络抖动/数据同步
                try {
                    TimeUnit.SECONDS.sleep(WAIT_SECONDS);
                } catch (InterruptedException ignored) {
                }
                regist();
            }
        });

    }

    private void regist() {
        System.out.printf("机器【%s】开始抢占 Master", port);
        System.out.println();
        try {
            zkClient.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(ROOT_PATH + "/master", port.getBytes());
            System.out.printf("机器【%s】成为了 Master", port);
            System.out.println();
            master = true;
        } catch (Exception e) {
            System.out.printf("机器【%s】抢占 Master 失败", port);
            System.out.println();
            master = false;
        }
    }
}

Controller：

@RestController
@RequestMapping("point")
public class PointController {

    @Autowired
    private MasterRegister masterRegister;

    @RequestMapping("master")
    public boolean isMaster() {
        return masterRegister.master;
    }
}

将项目打包，分别以不同的端口号 8080、8081、8082 启动：

➜  simple-spring-boot mvn clean package -Dmaven.test.skip=true
➜  ~ java -jar -Dserver.port=8080 /Users/dongguabai/Desktop/temp/simple-spring-boot/target/simple-spring-boot-0.0.1-SNAPSHOT.jar
➜  ~ java -jar -Dserver.port=8081 /Users/dongguabai/Desktop/temp/simple-spring-boot/target/simple-spring-boot-0.0.1-SNAPSHOT.jar
➜  ~ java -jar -Dserver.port=8082 /Users/dongguabai/Desktop/temp/simple-spring-boot/target/simple-spring-boot-0.0.1-SNAPSHOT.jar

控制台输出：

机器【8080】开始抢占 Master
机器【8080】成为了 Master

机器【8082】开始抢占 Master
机器【8082】抢占 Master 失败

机器【8081】开始抢占 Master
机器【8081】抢占 Master 失败

查看 ZooKeeper 节点情况：

[zk: 127.0.0.1:2181(CONNECTED) 10] ls /
[zookeeper, test]
[zk: 127.0.0.1:2181(CONNECTED) 11] ls /test
[master]
[zk: 127.0.0.1:2181(CONNECTED) 12] get /test/master
8080
cZxid = 0x19
ctime = Tue May 05 05:01:20 PDT 2020
mZxid = 0x19
mtime = Tue May 05 05:01:20 PDT 2020
pZxid = 0x19
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1000011f3160007
dataLength = 4
numChildren = 0
[zk: 127.0.0.1:2181(CONNECTED) 13] 

当前 Master 节点所在进程的端口号为 8080。

接下来关闭 8080 端口所在的进程，8081 和 8082 控制台分别输出：

节点变更，开始重新选举
机器【8081】开始抢占 Master
机器【8081】成为了 Master

节点变更，开始重新选举
机器【8082】开始抢占 Master
机器【8082】抢占 Master 失败

再看 ZooKeeper 节点信息：

[zk: 127.0.0.1:2181(CONNECTED) 54] get /test/master
8081
cZxid = 0x75
ctime = Tue May 05 05:36:50 PDT 2020
mZxid = 0x75
mtime = Tue May 05 05:36:50 PDT 2020
pZxid = 0x75
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1000011f3160020
dataLength = 4
numChildren = 0

可以看到当前 Master 节点所在进程的端口号为 8081。

第二种实现方式

第一种实现方式有一个问题就是会产生“惊群效应”，在并发量较高的情况下，也就是说短时间之内会有大量的客户端去争抢注册为 Master，短时间内会发生大量的事件上下文变更，但是实际上只有一个客户端可以注册得到，相当于出现了大量的无效的系统调度、上下文切换，系统系能大打折扣。为了解决这个问题，可以利用 ZooKeeper 中有序节点的特性。系统启动会在 /test/master 下注册临时有序节点，根据一定的策略去选定 Master 即可。

注册：

package com.example.simplespringboot.configuration;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @author Dongguabai
 * @Description
 * @Date 创建于 2020-05-05 18:54
 */
@Component
public class MasterRegister2 implements CommandLineRunner {

    private static final String ROOT_PATH = "/test";

    private static final Long WAIT_SECONDS = 3L;

    public volatile boolean master = false;

    public String masterPort;

    @Autowired
    private CuratorFramework zkClient;

    @Value("${server.port}")
    private String port;

    @Override
    public void run(String... args) throws Exception {
        //Spring 容器启动后创建临时节点
        regist();
    }

    private void regist() {
        try {
            zkClient.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(ROOT_PATH + "/master", port.getBytes());
        } catch (Exception ignored) {
        }
    }
}

将项目打包，分别以不同的端口号 8080、8081、8082 启动：

➜  simple-spring-boot mvn clean package -Dmaven.test.skip=true
➜  ~ java -jar -Dserver.port=8080 /Users/dongguabai/Desktop/temp/simple-spring-boot/target/simple-spring-boot-0.0.1-SNAPSHOT.jar
➜  ~ java -jar -Dserver.port=8081 /Users/dongguabai/Desktop/temp/simple-spring-boot/target/simple-spring-boot-0.0.1-SNAPSHOT.jar
➜  ~ java -jar -Dserver.port=8082 /Users/dongguabai/Desktop/temp/simple-spring-boot/target/simple-spring-boot-0.0.1-SNAPSHOT.jar

查看 ZooKeeper 节点：

[zk: 127.0.0.1:2181(CONNECTED) 57] ls /test
[master0000000019, master0000000017, master0000000018]

调用：

package com.example.simplespringboot.controller;

import com.example.simplespringboot.configuration.MasterRegister;
import com.example.simplespringboot.configuration.MasterRegister2;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Comparator;
import java.util.List;

/**
 * @author Dongguabai
 * @Description
 * @Date 创建于 2020-05-05 19:44
 */
@RestController
@RequestMapping("point")
public class PointController {

    @Autowired
    private CuratorFramework zkClient;

    @RequestMapping("master")
    public Object isMaster() throws Exception {
        //todo 缓存优化-监听变化
        List<String> list = zkClient.getChildren().forPath("/test");
        String s = list.stream().sorted(String.CASE_INSENSITIVE_ORDER).findFirst().get();
        return s;
    }
}

响应结果为：

master0000000017

关闭最早注册的机器后，再次访问，响应结果为：

master0000000018

References

• https://blog.csdn.net/qq_39833418/article/details/78316898
• https://blog.csdn.net/Dongguabai/article/details/82901686

欢迎关注公众号
​​​​​​