您的位置: 首页  >  文章  >  基于Apache Curator框架的ZooKeeper基本用法详解

基于Apache Curator框架的ZooKeeper基本用法详解

分类: 文章 • 2023-03-15 00:14:59

|  暂无评论 |  6495 views

一 简介

Apache Curator是一个比较完善的ZooKeeper客户端框架，通过封装的一套高级API 简化了ZooKeeper的操作。通过查看官方文档，可以发现Curator主要解决了三类问题：

  • 封装ZooKeeper client与ZooKeeper server之间的连接处理
  • 提供了一套Fluent风格的操作API
  • 提供ZooKeeper各种应用场景(recipe， 比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装

Curator主要从以下几个方面降低了zk使用的复杂性：

  • 重试机制:提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略，并且内部也提供了几种标准的重试策略(比如指数补偿)
  • 连接状态监控: Curator初始化之后会一直对zk连接进行监听，一旦发现连接状态发生变化将会作出相应的处理
  • zk客户端实例管理:Curator会对zk客户端到server集群的连接进行管理，并在需要的时候重建zk实例，保证与zk集群连接的可靠性
  • 各种使用场景支持:Curator实现了zk支持的大部分使用场景（甚至包括zk自身不支持的场景），这些实现都遵循了zk的最佳实践，并考虑了各种极端情况

二 基于Curator的ZooKeeper基本用法

首先需要在pom.xml中添加如下依赖：

 

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

        <!-- Apache Curator -->

        <dependency>

            <groupId>org.apache.curator</groupId>

            <artifactId>curator-framework</artifactId>

            <version>4.0.0</version>

        </dependency>

        <dependency>

            <groupId>org.apache.curator</groupId>

            <artifactId>curator-recipes</artifactId>

            <version>4.0.0</version>

        </dependency>

        <dependency>

            <groupId>org.apache.curator</groupId>

            <artifactId>curator-x-discovery</artifactId>

            <version>4.0.0</version>

        </dependency>

        <dependency>

            <groupId>org.apache.curator</groupId>

            <artifactId>curator-test</artifactId>

            <version>4.0.0</version>

            <scope>test</scope>

        </dependency>

 

（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

package cn.zifangsky.kafkademo.zookeeper;

 

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.apache.zookeeper.CreateMode;

 

import org.junit.Before;

import org.junit.Test;

 

/**

* 测试Apache Curator框架的基本用法

* API文档：http://curator.apache.org/apidocs/index.html

* @author zifangsky

*

*/

public class TestApacheCurator {

 

    //会话超时时间

    private final int SESSION_TIMEOUT = 30 * 1000;

    

    //连接超时时间

    private final int CONNECTION_TIMEOUT = 3 * 1000;

    

    //ZooKeeper服务地址

    private static final String SERVER = "192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102";

    

    //创建连接实例

    private CuratorFramework client = null;

    

    /**

     * baseSleepTimeMs：初始的重试等待时间

     * maxRetries：最多重试次数

     */

    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

    

    @Before

    public void init(){

        //创建 CuratorFrameworkImpl实例

        client = CuratorFrameworkFactory.newClient(SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, retryPolicy);

        

        //启动

        client.start();

    }

    

    /**

     * 测试创建节点

     * @throws Exception

     */

    @Test

    public void testCreate() throws Exception{

        //创建永久节点

        client.create().forPath("/curator","/curator data".getBytes());

        

        //创建永久有序节点

        client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/curator_sequential","/curator_sequential data".getBytes());

        

        //创建临时节点

        client.create().withMode(CreateMode.EPHEMERAL)

            .forPath("/curator/ephemeral","/curator/ephemeral data".getBytes());

 

        //创建临时有序节点

        client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)

            .forPath("/curator/ephemeral_path1","/curator/ephemeral_path1 data".getBytes());

        

        client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)

        .forPath("/curator/ephemeral_path2","/curator/ephemeral_path2 data".getBytes());

    }

 

}

在Curator中，首先需要通过CuratorFrameworkFactory创建连接实例，接着才能进行各种基本操作。需要说明的是，关于连接重试策略，Curator默认提供了以下几种：

  • ExponentialBackoffRetry：重试一定次数，每次重试时间依次递增
  • RetryNTimes：重试N次
  • RetryOneTime：重试一次
  • RetryUntilElapsed：重试一定时间

所谓永久节点，即：ZooKeeper中生成的节点名跟我们在代码中设置的节点名是一样的，而有序节点跟普通节点不同之处在于ZooKeeper会自动在生成的节点名后面添加***，具体效果如下：

基于Apache Curator框架的ZooKeeper基本用法详解

注：

临时节点，顾名思义，ZooKeeper会在节点生成一段时间之后自动删除该节点

此外，在创建临时有序节点的时候可以使用withProtection() 方法，该方法的作用是在创建的节点名前面添加GUID标识，其目的是为了避免出现“节点创建成功，但是ZooKeeper服务器在创建的节点名被返回给client前就出现了异常，从而导致临时节点没有被立即删除，而client也没法判断哪些节点被创建成功（注：创建的临时节点名有一定随机性）”的情况。具体效果如下：

基于Apache Curator框架的ZooKeeper基本用法详解

（2）检查某个节点是否存在：

在上面代码中添加以下单元测试：

 

1

2

3

4

5

6

7

8

9

10

11

12

    /**

     * 测试检查某个节点是否存在

     * @throws Exception

     */

    @Test

    public void testCheck() throws Exception{

        Stat stat1 = client.checkExists().forPath("/curator");

        Stat stat2 = client.checkExists().forPath("/curator2");

        

        System.out.println("'/curator'是否存在： " + (stat1 != null ? true : false));

        System.out.println("'/curator2'是否存在： " + (stat2 != null ? true : false));

    }

输出如下：

 

1

2

'/curator'是否存在： true

'/curator2'是否存在： false

 

（3）获取和设置节点数据：

继续添加以下单元测试：

 

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

    /**

     * 测试获取和设置节点数据

     * @throws Exception

     */

    @Test

    public void testGetAndSet() throws Exception{

        //获取某个节点的所有子节点

        System.out.println(client.getChildren().forPath("/"));

        

        //获取某个节点数据

        System.out.println(new String(client.getData().forPath("/curator")));

        

        //设置某个节点数据

        client.setData().forPath("/curator","/curator modified data".getBytes());

    }

输出如下：

 

1

2

[curator, curator_sequential0000000022, website, test, zookeeper, book]

/curator data

 

（4）异步设置节点数据以及获取通知：

继续添加以下单元测试：

 

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

    /**

     * 测试异步设置节点数据

     * @throws Exception

     */

    @Test

    public void testSetDataAsync() throws Exception{

        //创建监听器

        CuratorListener listener = new CuratorListener() {

            

            @Override

            public void eventReceived(CuratorFramework client, CuratorEvent event)

                    throws Exception {

                System.out.println(event.getPath());

            }

        };

        

        //添加监听器

        client.getCuratorListenable().addListener(listener);

        

        //异步设置某个节点数据

        client.setData().inBackground().forPath("/curator","/curator modified data with Async".getBytes());

        

        //为了防止单元测试结束从而看不到异步执行结果，因此暂停10秒

        Thread.sleep(10000);

    }

当然，除了上述方式之外还有另一种异步执行获取通知的方式：

 

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

    /**

     * 测试另一种异步执行获取通知的方式

     * @throws Exception

     */

    @Test

    public void testSetDataAsyncWithCallback() throws Exception{

        BackgroundCallback callback = new BackgroundCallback() {

            

            @Override

            public void processResult(CuratorFramework client, CuratorEvent event)

                    throws Exception {

                System.out.println(event.getPath());

            }

        };

    

        //异步设置某个节点数据

        client.setData().inBackground(callback).forPath("/curator","/curator modified data with Callback".getBytes());        

    

        //为了防止单元测试结束从而看不到异步执行结果，因此暂停10秒

        Thread.sleep(10000);    

    }

 

（5）删除节点：

继续添加以下单元测试：

 

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

    /**

     * 测试删除节点

     * @throws Exception

     */

    @Test

    public void testDelete() throws Exception{

        //创建测试节点

        client.create().orSetData().creatingParentContainersIfNeeded()

            .forPath("/curator/del_key1","/curator/del_key1 data".getBytes());

 

        client.create().orSetData().creatingParentContainersIfNeeded()

        .forPath("/curator/del_key2","/curator/del_key2 data".getBytes());

        

        client.create().forPath("/curator/del_key2/test_key","test_key data".getBytes());

        

        

        //删除该节点

        client.delete().forPath("/curator/del_key1");

        

        //级联删除子节点

        client.delete().guaranteed().deletingChildrenIfNeeded().forPath("/curator/del_key2");

    }

针对上面的单元测试，我简单说明一下：

  • orSetData()方法：如果节点存在则Curator将会使用给出的数据设置这个节点的值，相当于 setData() 方法
  • creatingParentContainersIfNeeded()方法：如果指定节点的父节点不存在，则Curator将会自动级联创建父节点
  • guaranteed()方法：如果服务端可能删除成功，但是client没有接收到删除成功的提示，Curator将会在后台持续尝试删除该节点
  • deletingChildrenIfNeeded()方法：如果待删除节点存在子节点，则Curator将会级联删除该节点的子节点

（6）事务管理：

继续添加以下单元测试：

 

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

    /**

     * 测试事务管理：碰到异常，事务会回滚

     * @throws Exception

     */

    @Test

    public void testTransaction() throws Exception{

        //定义几个基本操作

        CuratorOp createOp = client.transactionOp().create()

                .forPath("/curator/one_path","some data".getBytes());

        

        CuratorOp setDataOp = client.transactionOp().setData()

                .forPath("/curator","other data".getBytes());

        

        CuratorOp deleteOp = client.transactionOp().delete()

                .forPath("/curator");

        

        //事务执行结果

        List<CuratorTransactionResult> results = client.transaction()

                .forOperations(createOp,setDataOp,deleteOp);

        

        //遍历输出结果

        for(CuratorTransactionResult result : results){

            System.out.println("执行结果是： " + result.getForPath() + "--" + result.getType());

        }

    }

在上面的测试代码中，很显然因为节点“/curator”存在子节点，所以在删除的时候将会报错，以上代码并不能成功执行。这时观察ZKui节点，可以发现事务已经回滚了：

基于Apache Curator框架的ZooKeeper基本用法详解

（7）命名空间：

继续添加以下单元测试：

 

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

    /**

     * 测试命名空间

     * @throws Exception

     */

    @Test

    public void testNamespace() throws Exception{

        //创建带命名空间的连接实例

        CuratorFramework client2 = CuratorFrameworkFactory.builder()

                .namespace("mydemo/v1")

                .connectString(SERVER)

                .sessionTimeoutMs(SESSION_TIMEOUT)

                .connectionTimeoutMs(CONNECTION_TIMEOUT)

                .retryPolicy(retryPolicy)

                .build();

        //启动

        client2.start();

        

        client2.create().orSetData().creatingParentContainersIfNeeded()

        .forPath("/server1/method1","some data".getBytes());    

        

        client2.close();

    }

    

    @After

    public void close(){

        client.close();

    }

为了避免多个应用的节点名称冲突的情况，CuratorFramework提供了命名空间的概念。具体做法是：CuratorFramework会为它的API调用的节点路径的前面自动添加上命名空间。命名空间本质上是从根节点开始的一个路径，如：mydemo、mydemo/v1

运行上面代码之后，效果如下：

基于Apache Curator框架的ZooKeeper基本用法详解

参考：

相关推荐