Zookeeper：Zookeeper JavaAPI操作与分布式锁

一、Zookeeper JavaAPI操作

1、Curator介绍

• Curator是Apache Zookeeper的Java客户端。
• 常见的Zookeeper Java API：
  • 原生Java API。
  • ZkClient。
  • Curator。
• Curator项目目标是简化Zookeeper客户端的使用。
• Curator最初是Netfix研发的，后来捐献了Apache基金会，目前是Apache的顶级项目。
• 官网：https://curator.apache.org/docs/about

2、创建、查询、修改、删除节点

public class CuratorTest {
    private CuratorFramework client;

    @Before
    public void init() {
        // 1、方式一
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("localhost:2181", 60 * 1000, 15 * 1000, retryPolicy);

        // 2、方式二
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("localhost:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("test")
                .build();
        // 开启连接
        client.start();
        this.client = client;
    }

    /**
     * 1、基本创建：client.create().forPath("/app1")
     * 2、创建节点，带有数据：client.create().forPath("/app1", data)
     * 3、设置节点的类型: client.create().withMode(CreateMode.EPHEMERAL).forPath("/app1")
     * 4、创建多级节点: client.create().creatingParentsIfNeeded().forPath("/app1/app2")
     */
    @Test
    public void testCreate() {
        // 1、基本创建
        // 如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储
        try {
            String path = client.create().forPath("/app1");
            System.out.println(path);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

 /**
     * 查询节点：
     * 1、查询数据：get
     * 2、查询子节点: ls
     * 3、查询节点状态信息: ls -s
     */
    @Test
    public void testQueryData() {
        // 1、查询数据: get
        try {
            byte[] data = client.getData().forPath("/app1");
            System.out.println(new String(data));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void testQueryChildren() {
        // 查询子节点: ls
        try {
            List<String> stringList = client.getChildren().forPath("/");
            System.out.println(stringList);

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void testQueryState() {
        // 查询节点状态信息: ls -s
        Stat status = new Stat();
        try {
            client.getData().storingStatIn(status).forPath("/app1");
            System.out.println(status);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

  /**
     * 修改数据：
     * 1、修改数据。
     * 2、根据版本修改
     */
    @Test
    public void testSet() {
        // 修改数据
        try {
            client.setData().forPath("/app1", "haha".getBytes());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void testSetForVersion() throws Exception {
        // 根据版本修改
        Stat stat = new Stat();
        // 查询节点状态信息: ls -s
        client.getData().storingStatIn(stat).forPath("/app1");

        int version = stat.getVersion();
        client.setData().withVersion(version).forPath("/app1", "hehe".getBytes());
    }
 /**
     * 删除节点：delete、deleteall
     * 1、删除单个节点
     * 2、删除带有子节点的节点
     * 3、必须成功的删除: 为了防止网络抖动。本质就是重试。
     * 4、回调
     */
    @Test
    public void testSingleDelete() throws Exception {
        // 1、删除单个节点
        client.delete().forPath("/app1");

        // 2、删除带有子节点的节点
        client.delete().deletingChildrenIfNeeded().forPath("/app1");

        // 3、必须成功删除
        client.delete().guaranteed().forPath("/app1");

        //4、回调
        client.delete().guaranteed().inBackground((client, event) -> {
            System.out.println("我被删除了");
            System.out.println(event);
        }).forPath("/app1");
    }

    @After
    public void close() {
        if (client != null) {
            client.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
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

3、Watch事件监听

• Zookeeper允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，Zookeeper服务端会将事件通知到感兴趣的客户端上去，该机制是Zookeeper实现分布式协调服务的重要性。
• Zookeeper中引入了Watcher机制来实现了发布/订阅功能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。
• Zookeeper原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便需要开发人员自己反复注册Watcher，比较繁琐。
• Curator引入了Cache来实现对Zookeeper服务端事件的监听。
• Zookeeper提供了三种Watcher：
  • NodeCache：只是监听某一个特定的节点。
  • PathChildrenCache：监控一个ZNode的子节点。
  • TreeCache：可以监控整个树上所有节点，类似于PathChildrenCache和NodeCache的组合。

二、Zookeeper分布式锁原理