Apache Curator客户端的使用

布满荆棘的人生 2024-04-19 13:33 40阅读 0赞

## 当前已有的三种API客户端 ##

*  zk原生API :不支持超时自动重连，不支持节点递归创建
 *  zkclient：无文档，异常处理弱爆了(简单的抛出RuntimeException)
 *  Apache Curator：简化并提供各种场景的使用方法

### Apache Curator ###

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几个组成部分**

*  Client: 是ZooKeeper客户端的一个替代品, 提供了一些底层处理和相关的工具方法
 *  Framework: 用来简化ZooKeeper高级功能的使用, 并增加了一些新的功能, 比如管理到ZooKeeper集群的连接, 重试处理
 *  Recipes: 实现了通用ZooKeeper的recipe, 该组件建立在Framework的基础之上
 *  Utilities:各种ZooKeeper的工具类
 *  Errors: 异常处理, 连接, 恢复等
 *  Extensions: recipe扩展

## 使用 ##

pom.xml中添加如下依赖：

<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>

Java代码测试使用

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());
    	}
    
    	/**
    	 * 测试检查某个节点是否存在
    	 * @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));
    	}
    
    	/**
    	 * 测试获取和设置节点数据
    	 * @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());
    	}
    	/**
    	 * 测试异步设置节点数据
    	 * @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);
    	}
    
    	/**
    	 * 测试另一种异步执行获取通知的方式
    	 * @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);	
    	}
    
    	/**
    	 * 测试删除节点
    	 * @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");
    	}
    
    	/**
    	 * 测试事务管理：碰到异常，事务会回滚：
    	 * 在上面的测试代码中，很显然因为节点“/curator”存在子节点，所以在删除的时候将会报错，以上代码并不能成功执行。这时观察ZKui节点，可以发现事务已经回滚了：
    	 * @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());
    		}
    	}
    
    	/**
    	 * 测试命名空间
    	 * 为了避免多个应用的节点名称冲突的情况，CuratorFramework提供了命名空间的概念。具体做法是：CuratorFramework会为它的API调用的节点路径的前面自动添加上命名空间。命名空间本质上是从根节点开始的一个路径，如：mydemo、mydemo/v1
    	 * @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();
    	}
    
    }

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

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

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

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

### 具体场景使用 ###

多节点,多线程下发订单,使用zookeeper分布式锁机制保证订单正确接入oms系统  
https://blog.csdn.net/xiongxianze/article/details/79176000

## ZK UI客户端图形化查看工具 ##

zkui是github上的一个开源项目，主要实现了针对ZooKeepe的可视化增删改查功能。但是这个项目没有提供Release版本，需要自行下载源码编译。在这里，为了方便大家使用，我已经fork了这个项目，在修改了几处明显的bug之后编译了一个Pre-release版本，欢迎大家下载使用

下载地址：https://github.com/zifangsky/zkui/releases

（1）下载zkui-2.0.zip，并修改config.cfg配置文件：  
主要修改以下内容：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]  
（2）运行：  
Windows系统点击运行start.bat脚本，Linux系统则相应执行start.sh即可  
（3）访问WEB客户端并使用：  
访问：http://127.0.0.1:6060/  
同时使用 admin/admin 登录，登录之后的界面如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]

## 参考资料 ##

Apache Curator客户端的使用  
https://blog.csdn.net/weixin\_37778801/article/details/84704262  
https://blog.csdn.net/vbirdbest/article/details/82716934  
多节点,多线程下发订单,使用zookeeper分布式锁机制保证订单正确接入oms系统  
https://blog.csdn.net/xiongxianze/article/details/79176000

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/e4fd5251de5b42ac980280220c921cc7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/c97a83d4e94a40bcba21e95a9f877174.png