常见的zookeeper java api

  1. 原生java api, 难用

  2. ZkClient, 简单封装, 不太行

  3. Curator, netfix研发, 捐给了apache, 好用
    https://curator.apache.org/
    注意要对应zookeeper版本(我的是3.6.3), curator向下兼容, 所以我选择最新的稳定版本

    API

  4. 建立连接

  5. 添加节点

  6. 删除节点

  7. 修改节点

  8. 查询节点

  9. Watch事件监听

  10. 分布式锁实现

建立连接

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
package com.abc.curator.curator;

import com.sun.net.httpserver.Authenticator;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.junit.Test;

public class test {

    @Test
    public void testConnect() {

        //第一种方式
        /*
        1. ip:port
        2. 会话超时时间
        3. 连接超时时间
        4. 重试策略
         */
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
//        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.43.116:2181", 60 * 1000, 15 * 1000, retryPolicy);

//        client.start();

        //第二种方式
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.43.116:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy).namespace("/")
                .build();

        client.start();

    }
}

添加节点

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
package com.abc.curator.curator;

import com.sun.net.httpserver.Authenticator;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class test {
    private CuratorFramework client;

    @Before
    public void testConnect() {

        //第一种方式
        /*
        1. ip:port
        2. 会话超时时间
        3. 连接超时时间
        4. 重试策略
         */
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
//        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.43.116:2181", 60 * 1000, 15 * 1000, retryPolicy);

//        client.start();
        //开启

        //第二种方式
        client = CuratorFrameworkFactory.builder().connectString("192.168.43.116:2181")
                .sessionTimeoutMs(20 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("testzk")
                .build();

        client.start();
    }

    @Test
    public void testCreate() throws Exception {
        //如果没有指定数据, 那么默认把当前客户端的ip作为数据存储
        String s = client.create().forPath("/app1");
        System.out.println(s);

    }

    @After
    public void close() {
        if(client != null) {
            client.close();
        }
    }
}