devbox
空白诗007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Zookeeper Java 客户端 Curator 使用_zk 3.7.2 对应的curator-recipes版本

作者：空白诗007 | 2024-08-05 19:25:49

赞

踩

zk 3.7.2 对应的curator-recipes版本

Zookeeper 3.6.1

添加依赖

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

  • 1
  • 2
  • 3
  • 4
  • 5

添加配置

zookeeper:
  # 连接，多个用逗号隔开
  connect-string: 192.168.110.40:2181
  # 连接超时时间
  connection-timeout-ms: 5000
  # 会话超时时间
  session-timeout-ms: 60000
  # 初始休眠时间
  base-sleep-time-ms: 1000
  # 最大重试次数
  max-retries: 3

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

初始化

@Configuration
public class ZookeeperConfig {

    @Value("${zookeeper.connect-string:''}")
    private String connectString;

    @Value("${zookeeper.connection-timeout-ms:''}")
    private int connectionTimeoutMs;

    @Value("${zookeeper.session-timeout-ms:''}")
    private int sessionTimeoutMs;

    @Value("${zookeeper.base-sleep-time-ms:''}")
    private int baseSleepTimeMs;

    @Value("${zookeeper.max-retries:''}")
    private int maxRetries;

    @Bean(destroyMethod = "close")
    public CuratorFramework getClient() {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString(connectString)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .retryPolicy(new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries))
                .build();
        client.start();
        return client;
    }
}

  • 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

测试

@SpringBootTest
class CuratorTestApplicationTests {

    @Autowired
    private CuratorFramework client;

    /**
     * 创建节点
     * @throws Exception
     */
    @Test
    public void createNode() throws Exception {
        // 非递归创建
        // 持久节点
        client.create().forPath("/persistent_path", "/persistent_path data".getBytes(StandardCharsets.UTF_8));

        // 持久顺序节点
        client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/persistent_path/seq_");
        client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/persistent_path/seq_");
        client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/persistent_path/seq_");

        // 临时节点
        client.create().withMode(CreateMode.EPHEMERAL).forPath("/ephemeral_path", "/ephemeral_path data".getBytes(StandardCharsets.UTF_8));

        // 临时顺序节点
        client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/persistent_path/eph_seq_");
        client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/persistent_path/eph_seq_");
        client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/persistent_path/eph_seq_");

        // sleep一段时间，验证临时节点
        Thread.sleep(60000);

        // 递归创建
        // creatingParentContainersIfNeeded() 会递归创建一个容器类型的父节点，如果当前版本不支持容器类型的父节点，则会创建一个普通的持节化节点。
        // 当容器类型的节点下的子节点全部被删除时，则该节点也会在一段时间内被删除。
        client.create().creatingParentContainersIfNeeded().forPath("/persistent_path_2/path_1", "/persistent_path_2/path_1 data".getBytes(StandardCharsets.UTF_8));
        client.create().creatingParentContainersIfNeeded().forPath("/persistent_path_2/path_2", "/persistent_path_2/path_2 data".getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 节点是否存在，返回节点状态
     * @throws Exception
     */
    @Test
    public void exist() throws Exception {
        Stat stat = client.checkExists().forPath("/persistent_path");
        System.out.println(stat);
    }

    /**
     * 获取数据
     * @throws Exception
     */
    @Test
    public void getData() throws Exception {
        String data = new String(client.getData().forPath("/persistent_path"), StandardCharsets.UTF_8);
        System.out.println(data);
    }

    /**
     * 更新数据
     * @throws Exception
     */
    @Test
    public void setData() throws Exception {
        client.setData().forPath("/persistent_path", "/persistent_path update data".getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 获取子节点
     * @throws Exception
     */
    @Test
    public void getChildren() throws Exception {
        List<String> childrenList = client.getChildren().forPath("/persistent_path");
        System.out.println(childrenList);
    }

    /**
     * 删除节点
     */
    @Test
    public void deleteNode() throws Exception {
        // 非递归删除
        client.delete().forPath("/persistent_path_2/path_1");
        // guaranteed() 保证在session有效的情况下，后台持续进行该节点的删除操作，直到删除掉
        client.delete().guaranteed().forPath("/persistent_path_2/path_2");

        // 递归删除当前节点及其子节点
        client.delete().deletingChildrenIfNeeded().forPath("/persistent_path");
    }

    /**
     * 事务
     * @throws Exception
     */
    @Test
    public void transaction() throws Exception {
        CuratorOp createOp = client.transactionOp().create().forPath("/persistent_path/path", "/persistent_path/path data".getBytes());
        CuratorOp setDataOp = client.transactionOp().setData().forPath("/persistent_path/path", "/persistent_path/path update data".getBytes());
        CuratorOp deleteOp = client.transactionOp().delete().forPath("/persistent_path/path");

        List<CuratorTransactionResult> resultList = client.transaction().forOperations(createOp, setDataOp, deleteOp);
        for(CuratorTransactionResult result : resultList) {
            System.out.println(result.getForPath() + " - " + result.getType());
        }
    }
}

  • 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

测试事件

/**
 * 注册一次性监听事件
 * @throws Exception
 */
@Test
public void oneTimeWatcher() throws Exception {
    byte[] bytes = client.getData().usingWatcher((Watcher) watchedEvent -> {
        System.out.println("watchedEvent " + watchedEvent);
    }).forPath("/persistent_path");
    String data = new String(bytes, StandardCharsets.UTF_8);

    System.out.println("节点 /persistent_path 内容：" + data);
    // 第一次变更节点数据触发
    client.setData().forPath("/persistent_path","/persistent_path data update 1".getBytes());
    // 第二次变更节点数据没触发
    client.setData().forPath("/persistent_path","/persistent_path data update 2".getBytes());
}

/**
 * 针对 background 通知和错误通知。
 * 使用CuratorListener后，调用 inBackground() 方法会异步获得监听，对于节点的创建或修改则不会触发监听事件。
 * @throws Exception
 */
@Test
public void curatorListener() throws Exception {
    CuratorListener listener = (client, event) -> {
        System.out.println("event：" + event);
    };

    client.getCuratorListenable().addListener(listener);
    // 异步获取数据
    client.getData().inBackground().forPath("/persistent_path");
}

  • 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

测试 Cache 事件

/**
 * 单独监听当前节点下所有子节点增删改
 * @throws Exception
 */
@Test
public void cache() throws Exception {
    String path = "/persistent_path";
    CuratorCache curatorCache = CuratorCache.build(client, path);

    CuratorCacheListener listener = CuratorCacheListener.builder()
            .forInitialized(() -> {
                System.out.println("初始化");
            })
            .forCreates(childData -> {
                System.out.println("添加节点：" + childData.getPath());
            })
            .forChanges(((oldNode, node) -> {
                System.out.println("更新节点：" + oldNode.getPath());
            }))
            .forDeletes(childData -> {
                System.out.println("删除节点：" + childData.getPath());
            })
            .build();
    curatorCache.listenable().addListener(listener);
    curatorCache.start();

    // 添加节点
    client.create().forPath("/persistent_path/path_1");
    client.create().forPath("/persistent_path/path_1/path_1");

    Thread.sleep(1000);

    // 更新节点
    client.setData().forPath("/persistent_path/path_1", "/persistent_path/path_1 data".getBytes());
    client.setData().forPath("/persistent_path/path_1/path_1", "/persistent_path/path_1/path_1 data".getBytes());

    Thread.sleep(1000);

    // 删除节点
    client.delete().forPath("/persistent_path/path_1/path_1");
    client.delete().forPath("/persistent_path/path_1");
}

/**
 * 监听节点数据增删改
 */
@Test
public void nodeCache() throws Exception {
    String path = "/persistent_path";
    CuratorCache curatorCache = CuratorCache.build(client, path);

    CuratorCacheListener listener = CuratorCacheListener.builder()
            .forNodeCache(() -> {
                if (curatorCache.get(path).isPresent()) {
                    System.out.print("节点 " + curatorCache.get(path).get().getPath());
                    System.out.println(" 数据变化：" + new String(curatorCache.get(path).get().getData(), StandardCharsets.UTF_8));
                }
            }).build();
    curatorCache.listenable().addListener(listener);
    curatorCache.start();

    client.setData().forPath(path, (path + " update data 1").getBytes(StandardCharsets.UTF_8));
    client.setData().forPath(path, (path + " update data 2").getBytes(StandardCharsets.UTF_8));
    client.setData().forPath(path, (path + " update data 3").getBytes(StandardCharsets.UTF_8));
}

/**
 * 监听一级子节点增删改
 */
@Test
public void pathChildrenCache() throws Exception {
    String path = "/persistent_path";
    CuratorCache curatorCache = CuratorCache.build(client, path);
    CuratorCacheListener listener = CuratorCacheListener.builder()
            .forPathChildrenCache(path, client, (client, event) -> {
                switch (event.getType()){
                    case CHILD_ADDED:
                        System.out.println("添加子节点：" + event.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        System.out.println("更新子节点：" + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        System.out.println("删除子节点：" + event.getData().getPath());
                        break;
                    default:
                }
            })
            .build();
    curatorCache.listenable().addListener(listener);
    curatorCache.start();

    // 添加子节点
    client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/persistent_path/seq_");
    client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/persistent_path/seq_");

    // 更新子节点
    String firstNode = client.getChildren().forPath(path).get(0);
    client.setData().forPath(path + "/" + firstNode, (path + "/" + firstNode).getBytes());

    // 删除子节点
    String secondNode = client.getChildren().forPath(path).get(1);
    client.delete().forPath(path + "/" + secondNode);
}

/**
 * 监听当前节点下所有子节点增删改
 */
@Test
public void treeNodeCache() throws Exception {
    String path = "/persistent_path";
    CuratorCache curatorCache = CuratorCache.build(client, path);
    CuratorCacheListener listener = CuratorCacheListener.builder()
            .forTreeCache(client, (client, event) -> {
                switch (event.getType()){
                    case NODE_ADDED:
                        System.out.println("添加节点：" + event.getData().getPath());
                        break;
                    case NODE_UPDATED:
                        System.out.println("更新节点：" + event.getData().getPath());
                        break;
                    case NODE_REMOVED:
                        System.out.println("删除节点：" + event.getData().getPath());
                        break;
                    default:
                }
            })
            .build();
    curatorCache.listenable().addListener(listener);
    curatorCache.start();

    // 添加节点
    client.create().forPath("/persistent_path/path_1");
    client.create().forPath("/persistent_path/path_1/path_1");

    // 更新节点
    client.setData().forPath("/persistent_path/path_1", "/persistent_path/path_1 data".getBytes());
    client.setData().forPath("/persistent_path/path_1/path_1", "/persistent_path/path_1/path_1 data".getBytes());

    // 删除节点
    client.delete().forPath("/persistent_path/path_1/path_1");
    client.delete().forPath("/persistent_path/path_1");
}

  • 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
  • 139
  • 140
  • 141
  • 142
  • 143

测试异步

/**
 * 异步，可指定线程池
 */
@Test
public void async() throws Exception {
    client.getData().inBackground((client, event) -> {
        System.out.println(Thread.currentThread().getName() + " " + event);
    }).forPath("/persistent_path");
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

测试分布式锁

/**
 * 排它锁
 */
@Test
public void lock() throws Exception {
    InterProcessMutex lock = new InterProcessMutex(client, "/lock");

    Thread.sleep(10000);
    System.out.println("等待获取锁");
    lock.acquire();

    System.out.println("获取锁");
    Thread.sleep(10000);

    System.out.println("释放锁");
    lock.release();
}

/**
 * 读锁
 */
@Test
public void readLock() throws Exception {
    InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/readLock");
    InterProcessMutex readLock = lock.readLock();

    Thread.sleep(10000);
    System.out.println("等待获取读锁");
    readLock.acquire();

    System.out.println("获取读锁");
    Thread.sleep(10000);

    System.out.println("释放读锁");
    readLock.release();
}

/**
 * 写锁
 */
@Test
public void writeLock() throws Exception {
    InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/writeLock");
    InterProcessMutex writeLock = lock.writeLock();

    Thread.sleep(10000);
    System.out.println("等待获取写锁");
    writeLock.acquire();

    System.out.println("获取写锁");
    Thread.sleep(10000);

    System.out.println("释放锁锁");
    writeLock.release();
}

  • 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

参考：
Apache Curator
apache/curator
聊一聊 Zookeeper 客户端之 Curator
Curator Api使用

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/空白诗007/article/detail/933841
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号