Zookeeper：实现“命名服务”的 Demo

命名服务是分步实现系统中较为常见的一类场景，分布式系统中，被命名的实体通常可以是集群中的机器、提供的服务地址或远程对象等，通过命名服务，客户端可以根据指定名字来获取资源的实体、服务地址和提供者的信息。Zookeeper 也可帮助应用系统通过资源引用的方式来实现对资源的定位和使用，广义上的命名服务的资源定位都不是真正意义上的实体资源，在分布式环境中，上层应用仅仅需要一个全局唯一的名字。Zookeeper 可以实现一套分布式全局唯一 ID 的分配机制。

～

本篇内容包括：Demo 概述、代码实现、测试结果

---

---

一、Demo 概述

1、关于 zookeeper “命名服务”

分布式命名服务，创建一个节点后，节点的路径就是全局唯一的，可以作为全局名称使用

命名服务是分步实现系统中较为常见的一类场景，分布式系统中，被命名的实体通常可以是集群中的机器、提供的服务地址或远程对象等，通过命名服务，客户端可以根据指定名字来获取资源的实体、服务地址和提供者的信息。Zookeeper 也可帮助应用系统通过资源引用的方式来实现对资源的定位和使用，广义上的命名服务的资源定位都不是真正意义上的实体资源，在分布式环境中，上层应用仅仅需要一个全局唯一的名字。Zookeeper 可以实现一套分布式全局唯一 ID 的分配机制。

2、Demo 设计

Zookeeper 中的 ZNode 数据一共有 4 种：持久节点、持久顺序节点、临时节点、临时顺序节点

我们通过利用 zookeeper 创建持久顺序节点和临时顺序节点，则可天然的为我们创建有前缀含义的、自增的、唯一的分布式主键：

1. 创建持久的顺序节点
2. 获取返回的节点名称，提取ID
3. 按需删除节点（防止创建大量的持久节点）

3、Demo 前提

参考：Mac通过Docker安装Zookeeper集群

---

二、代码实现

1、引用 Maven 依赖

        
        <dependency>
            <groupId>org.apache.zookeepergroupId>
            <artifactId>zookeeperartifactId>
            <version>3.7.0version>
        dependency>
1
2
3
4
5
6

2、ConnectionWatcher 类创建 Zookeeper 连接

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

public class ConnectionWatcher implements Watcher {

    private final CountDownLatch connectedSignal = new CountDownLatch(1);
    private static final int SESSION_TIMEOUT = 5000;
    protected ZooKeeper zk;


    public void connect(String hosts) throws IOException, InterruptedException {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
        connectedSignal.await();
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState() == Event.KeeperState.SyncConnected) {
            connectedSignal.countDown();
        }
    }

    public void close() throws InterruptedException {
        zk.close();
    }

}
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

3、ActiveKeyValueStore 类读写 Zookeeper 数据

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

public class ActiveKeyValueStore extends ConnectionWatcher {

    private static final Charset CHARSET = StandardCharsets.UTF_8;

    /**
     * 注册临时有序节点
     *
     * @param path  节点地址
     * @param value 数据值
     * @throws InterruptedException 中断异常
     * @throws KeeperException      ZooKeeper异常
     */
    public void writeEphemeralSequentialNode(String path, String value) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(path, false);
        if (stat == null) {
            if (value == null) {
                // EPHEMERAL_SEQUENTIAL-临时有序，PERSISTENT_SEQUENTIAL-持久有序
                zk.create(path, null,
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            } else {
                zk.create(path, value.getBytes(CHARSET),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            }

        } else {
            if (value == null) {
                zk.setData(path, null, -1);
            } else {
                zk.setData(path, value.getBytes(CHARSET), -1);
            }
        }
    }

    /**
     * 获取所有子节点
     *
     * @param path    节点地址
     * @param watcher watcher
     * @return 所有子节点
     * @throws InterruptedException 中断异常
     * @throws KeeperException      ZooKeeper异常
     */
    public List<String> readChildren(String path, Watcher watcher) throws InterruptedException, KeeperException {
        List<String> childrens = null;
        if (watcher == null) {
            childrens = zk.getChildren(path, false);
        } else {
            childrens = zk.getChildren(path, watcher, null);
        }
        return childrens;
    }
}
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

4、ConfigWatcher 类自定义 Watcher

public class ConfigWatcher implements Watcher {
  
   public ConfigWatcher() {}
  
    @Override
    public void process(WatchedEvent event) {}

}
1
2
3
4
5
6
7
8

5、IdMaker 类实现命名服务

import org.apache.zookeeper.KeeperException;

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

public class IdMaker {

    private final ActiveKeyValueStore store;

    public IdMaker(String hosts) throws IOException, InterruptedException {
        //定义一个类
        store = new ActiveKeyValueStore();
        //连接Zookeeper
        store.connect(hosts);
    }

    public void creatNode() throws InterruptedException, KeeperException {

        String writeEphemeralPath = "/configuration/my_demo_id_";

        store.writeEphemeralSequentialNode(writeEphemeralPath, "测试有序节点1");
        store.writeEphemeralSequentialNode(writeEphemeralPath, "测试有序节点2");
        store.writeEphemeralSequentialNode(writeEphemeralPath, "测试有序节点3");

    }

    public List<String> readChildrenNode(String hosts) throws IOException, InterruptedException, KeeperException {
        String path = "/configuration";
        //创建 watcher
        ConfigWatcher watcher = new ConfigWatcher();
        return store.readChildren(path, watcher);
    }


    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

        String hosts = "localhost:2181";

        IdMaker idMaker = new IdMaker(hosts);
        idMaker.creatNode();
        List<String> nodes = idMaker.readChildrenNode(hosts);
        for (String s : nodes) {
            System.out.println(s);
        }
    }

}
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

---

三、测试结果

IdMaker 代码测试结果如下：

my_demo_id_0000000001
my_demo_id_0000000002
my_demo_id_0000000003
1
2
3

通过 IdMaker 打印的信息可以看出，已经成功模拟实现命名服务协调