Apache Curator

不念不忘少年蓝@ 2023-01-21 14:17 65阅读 0赞

Apache Curator 是Apache ZooKeeper 的Java 客户端库。Curator 项目的目标是简化ZooKeeper 客户端的使用。例如，在以前的代码展示中，我们都要自己处理  
ConnectionLossException 。另外Curator 为常见的分布式协同服务提供了高质量的实现。Apache Curator 最初是Netflix 研发的，后来捐献给了Apache 基金会，目前是Apache 的顶级项目。

### Curator 技术栈 ###

**Client**：封装了 Zookeeper 类，管理和Zookeeper集群的连接，并提供了重建连接机制。

**Framework**：为所有的Zookeeper操作提供了重试机制，对提供了一个Fluent风格的API。

**Recipes**：使用framework实现了大量 Zookeeper 协同服务。

**Extensions**：扩展模块

![image-20210516170157518][]

Client 的创建，初始化一个client分为两步骤：

1.  创建client
2.  启动client

//创建 client 
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    
    // 使用 Factory 方法
    CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
    // Fluent 风格
    CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString(connectString)
            .retryPolicy(retryPolicy)
       	    .build();	
    //启动 client 
    client.start();

**Fluent 风格API**

// 同步版本
    client.create().withMode(CreateMode.PERSISTENT).forPath(path, data);
    // 异步版本
    client.create().withMode(CreateMode.PERSISTENT).inBackground().forPath(path, data);
    // 使用watch
    client.getData().watched().forPath(path);

案例：

public class TestCurator { 
        private CuratorFramework client = null;
        private String connectString = "192.168.1.100:2181";
        private RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        @Before
        public void creatClient() { 
            client = CuratorFrameworkFactory.builder()
                    .connectString(connectString)
                    .retryPolicy(retryPolicy)
                    .build();
            client.start();
        }
    
        @Test
        public void testSyncOp() throws Exception { 
            String path = "/one";
            byte[] data = "testSyncOp".getBytes();
            client.create().withMode(CreateMode.PERSISTENT).forPath(path,data);
    
            byte[] actualData = client.getData().forPath(path);
            System.out.println(new String(actualData));
            client.delete().forPath(path);
            client.close();
        }
    
        @Test
        public void testAsyncOp() throws Exception { 
            String path = "/two";
            final byte[] data = "testAsyncOp".getBytes();
            final CountDownLatch latch = new CountDownLatch(1);
    
            client.getCuratorListenable()
                    .addListener((CuratorFramework c, CuratorEvent event) -> { 
                        switch (event.getType()) { 
                            case CREATE:
                                System.out.printf("znode '%s' created\n", event.getPath());
                                c.getData().inBackground().forPath(event.getPath());
                                break;
                            case GET_DATA:
                                System.out.printf("got the data of znode '%s'\n", event.getPath());
                                System.out.println(new String(event.getData()));
                                c.delete().inBackground().forPath(path);
                                break;
                            case DELETE:
                                System.out.printf("znode '%s' deleted.\n", event.getPath());
                                latch.countDown();
                                break;
                        }
                    });
    
            client.create().withMode(CreateMode.PERSISTENT).inBackground().forPath(path, data);
    
            latch.await();
    
            client.close();
        }
    
        @Test
        public void testWatch() throws Exception { 
            String path = "/three";
            byte[] data = { '3'};
            byte[] newData = { '4'};
            CountDownLatch latch = new CountDownLatch(1);
    
            client.getCuratorListenable()
                    .addListener(((CuratorFramework c, CuratorEvent event ) -> { 
                        switch (event.getType()) { 
                            case WATCHED:
                                WatchedEvent we = event.getWatchedEvent();
                                System.out.println("watched event:" + we);
                                if (we.getType() == Watcher.Event.EventType.NodeDataChanged
                                    && we.getPath().equals(path)) { 
    
                                    System.out.println("got the event for the triggered watch");
                                    byte[] actualData = c.getData().forPath(path);
                                    System.out.println(new String(actualData));
                                }
                                latch.countDown();
                                break;
                        }
                    }));
            client.create().withMode(CreateMode.PERSISTENT).forPath(path, data);
            byte[] actualData = client.getData().watched().forPath(path);
            System.out.println(new String(actualData));
    
            client.setData().forPath(path, newData);
            latch.await();
    
            client.delete().forPath(path);
        }
    
        @Test
        public void testCallbackAndWatch() throws Exception { 
            String path = "/four";
            byte[] data = { '4'};
            byte[] newData = { '5'};
            CountDownLatch latch = new CountDownLatch(3);
    
            client.getCuratorListenable()
                    .addListener(((CuratorFramework c, CuratorEvent event ) -> { 
                        switch (event.getType()) { 
                            case CREATE:
                                System.out.printf("znode '%s' created\n", event.getPath());
                                System.out.println(new String(client.getData().watched().forPath(path)));
                                client.setData().forPath(path, newData);
                                latch.countDown();
                                break;
                            case WATCHED:
                                WatchedEvent we = event.getWatchedEvent();
                                System.out.println("watched event:" + we);
                                if (we.getType() == Watcher.Event.EventType.NodeDataChanged
                                        && we.getPath().equals(path)) { 
    
                                    System.out.println("got the event for the triggered watch");
                                    byte[] actualData = c.getData().forPath(path);
                                    System.out.println(new String(actualData));
                                }
                                latch.countDown();
                                break;
                        }
                    }));
            client.create().withMode(CreateMode.PERSISTENT).inBackground().forPath(path, data);
            latch.await();
    
            client.delete().forPath(path);
        }
    }

参考链接：https://time.geekbang.org/course/intro/100034201

[image-20210516170157518]: /images/20221020/a78dbf6661174531be5e2beea04a397c.png