4、zookeeper的java三种客户端-Curatorcrud、事务操作、监听、分布式计数器、分布式锁

Zookeeper系列文章目录

1、zookeeper3.7.1安装与验证
2、zookeeper基本操作及应用示例（shell、java api、应用场景示例）
3、zookeeper的选举----经验证符合事实，网上很多都是错误的
4、zookeeper的java三种客户端介绍-Curator（crud、事务操作、监听、分布式计数器、分布式锁）
5、zookeeper的java -Curator（服务注册与发现）

---

---

zookeeper常用的3种java客户端

• zookeeper原生Java API
• ZkClient
• Apache curator
  本文将按照顺序逐一简单介绍其使用。

一、zookeeper原生Java API

Zookeeper客户端提供了基本的操作，比如创建会话、创建节点、读取节点、更新数据、删除节点和检查节点是否存在等。但对于开发人员来说，Zookeeper提供的基本操纵还是有一些不足之处。
Zookeeper API不足之处
（1）Session超时之后没有实现重连机制，需要手动操作；
（2）Watcher注册是一次性的，每次触发之后都需要重新进行注册；
（3）不支持递归创建节点；
（4）异常处理繁琐，Zookeeper提供了很多异常，对于开发人员来说可能根本不知道该如何处理这些异常信息；
（5）只提供了简单的byte[]数组的接口，没有提供针对对象级别的序列化；
（6）创建节点时如果节点存在抛出异常，需要自行检查节点是否存在；
（7）删除节点无法实现级联删除；
基于以上原因，直接使用Zookeeper原生API的人并不多。

具体使用参见文章zookeeper的基本操作及应用示例中的介绍。

二、ZkClient

ZkClient是一个开源客户端，在Zookeeper原生API接口的基础上进行了包装，更便于开发人员使用。解决如下问题：
1）session会话超时重连
2）解决Watcher反复注册
3）简化API开发
虽然 ZkClient 对原生 API 进行了封装，但也有它自身的不足之处：

• 几乎没有参考文档；
• 异常处理简化（抛出RuntimeException）；
• 重试机制比较难用；
• 没有提供各种使用场景的实现；

示例欠奉，pom.xml文件增加如下内容：

<!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>

三、Apache curator

Curator是Netflix公司开源的一套Zookeeper客户端框架，和ZkClient一样，解决了非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等。目前已经成为 Apache 的顶级项目。
官网：http://curator.apache.org/index.html
Apache Curator是Apache ZooKeeper的Java / JVM客户端库，Apache ZooKeeper是一种分布式协调服务。它包括一个高级API框架和实用程序，使Apache ZooKeeper更容易和更可靠。它还包括常见用例和扩展（如服务发现和Java 8异步DSL）的配方。
其特点：

• Apache 的开源项目
• 解决Watch注册一次就会失效的问题
• 提供一套Fluent风格的 API 更加简单易用
• 提供更多解决方案并且实现简单，例如：分布式锁
• 提供常用的ZooKeeper工具类
• 编程风格更舒服

除此之外，Curator中还提供了Zookeeper各种应用场景（Recipe，如共享锁服务、Master选举机制和分布式计算器等）的抽象封装。

Curator项目组件

组件的Maven依赖

源码地址：https://github.com/apache/curator
如果只想使用Curator操作Zookeeper增删改查，则使用curator-client包及curator-framework包。zookeeper的版本是3.7.1

1、pom.xml

<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-framework -->
<dependency>
	<groupId>org.apache.curator</groupId>
	<artifactId>curator-framework</artifactId>
	<version>5.3.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-client -->
<dependency>
	<groupId>org.apache.curator</groupId>
	<artifactId>curator-client</artifactId>
	<version>5.3.0</version>
</dependency>

2、定义常量类

import org.apache.curator.retry.ExponentialBackoffRetry;

/**
 * @author alanchan
 *
 */
public class Constant {
	// 会话超时时间 10000
	public final static int SESSION_TIMEOUT = 10 * 1000;

	// 连接超时时间 50000
	public final static int CONNECTION_TIMEOUT = 50 * 1000;

	// ZooKeeper服务地址
	public static String zkServerAddress = "192.168.10.41:2118,192.168.10.42:2118,192.168.10.43:2118";

	// 1 重试策略：初试时间为1s 重试3次
	public static ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
}

3、连接实例化

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.curator.utils.CloseableUtils;
import org.apache.zookeeper.ZooKeeper.States;


public class App {

	public static void main(String[] args) {
		// 例子1
//		test1();

		// 例子2
//		CuratorFramework zkClient = createSimple(zkServerAddress);
//		zkClient.start();

		// 例子3
		ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
		CuratorFramework zkClient = createWithOptions(Constant.zkServerAddress, retryPolicy, 5000, 3000);
		zkClient.start();

		System.out.println(States.CONNECTED);
		System.out.println(zkClient.getState());
	}

	public static void test1() {
//    * @param baseSleepTimeMs initial amount of time to wait between retries  初始重试等待时间
//    * @param maxRetries max number of times to retry 最大重试次数
//    * @param maxSleepMs max time in ms to sleep on each retry 最大重试等待时间
		ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);

		CuratorFramework zkClient = CuratorFrameworkFactory.builder().connectString(Constant.zkServerAddress)
				.sessionTimeoutMs(3000).connectionTimeoutMs(5000).retryPolicy(retryPolicy).build();
		// 很重要 一定要调用start来创建session链接
		zkClient.start();

		CloseableUtils.closeQuietly(zkClient);
	}

	public static CuratorFramework createSimple(String connectionString) {
		// these are reasonable arguments for the ExponentialBackoffRetry. The first
		// retry will wait 1 second - the second will wait up to 2 seconds - the
		// third will wait up to 4 seconds.
		ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);

		// The simplest way to get a CuratorFramework instance. This will use default
		// values.
		// The only required arguments are the connection string and the retry policy
		return CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
	}

	// int connectionTimeoutMs  50000
	// int sessionTimeoutMs 10000
	//连接时间要大于会话时间，如果设置不当，会出现 KeeperErrorCode = ConnectionLoss异常
	public static CuratorFramework createWithOptions(String connectionString, RetryPolicy retryPolicy,
			int connectionTimeoutMs, int sessionTimeoutMs) {
		// using the CuratorFrameworkFactory.builder() gives fine grained control
		// over creation options. See the CuratorFrameworkFactory.Builder javadoc
		// details
		return CuratorFrameworkFactory.builder().connectString(connectionString).retryPolicy(retryPolicy)
				.connectionTimeoutMs(connectionTimeoutMs).sessionTimeoutMs(sessionTimeoutMs)
				// etc. etc.
				.build();
	}
}

4、事务操作示例

import java.util.Collection;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.zookeeper_curator.App;
import org.zookeeper_curator.Constant;

/**
 * @author alanchan Curator 事务管理 操作
 */
public class TransactionExamples {

	public static void main(String[] args) throws Exception {

		transaction(getCuratorFramework());

	}

	private static CuratorFramework getCuratorFramework() {
		CuratorFramework client = App.createWithOptions(Constant.zkServerAddress, Constant.retryPolicy,
				Constant.CONNECTION_TIMEOUT, Constant.SESSION_TIMEOUT);
		client.start();
		return client;
	}

	public static Collection<CuratorTransactionResult> transaction(CuratorFramework client) throws Exception {
		// this example shows how to use ZooKeeper's transactions

		CuratorOp createOp = client.transactionOp().create().forPath("/a/path", "some data".getBytes());
		CuratorOp setDataOp = client.transactionOp().setData().forPath("/another/path", "other data".getBytes());
		CuratorOp deleteOp = client.transactionOp().delete().forPath("/yet/another/path");

		// 将上述基本操作封装程一个事务
		//如果上述三个操作都没有事先创建路径的话，不会成功任何一个命令
		//为了验证，先创建一个/a目录，按照不增加事务控制的话，create /a/path，应该成功，因为增加了事务控制，应该都不会成功
		//如果先创建了/a目录，第一条命令create /a/path会成功，但没有提前创建/another/path，所以修改数据不会成功
		Collection<CuratorTransactionResult> results = client.transaction().forOperations(createOp, setDataOp,
				deleteOp);

		for (CuratorTransactionResult result : results) {
			System.out.println(result.getForPath()   " - "   result.getType());
		}

		return results;
	}
}

5、CRUD示例

import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.zookeeper_curator.App;
import org.zookeeper_curator.Constant;

/**
 * @author alanchan
 *
 */
public class CrudExamples {
	// 创建连接实例
	private static CuratorFramework zkClient = null;

	public static void main(String[] args) throws Exception {

		// int connectionTimeoutMs 50000
		// int sessionTimeoutMs 10000
		// 2 通过工厂创建连接
		zkClient = App.createWithOptions(Constant.zkServerAddress, Constant.retryPolicy, Constant.CONNECTION_TIMEOUT,
				Constant.SESSION_TIMEOUT);

		zkClient.start();

//		zkClient.create().forPath("/testcurator", "testing".getBytes());

//		create(zkClient, "/testcurator", "testing".getBytes());
//		createEphemeral(zkClient, "/testcurator", "testing".getBytes());

//		createEphemeralSequential(zkClient, "/testcurator", "testing".getBytes());
//		createIdempotent(zkClient, "/testcurator", "testing".getBytes());

//		setData(zkClient, "/testcurator", "testing2".getBytes());

//		setDataAsync(zkClient, "/testcurator", "testing32".getBytes());

		BackgroundCallback callback = new BackgroundCallback() {

			@Override
			public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
				System.out.println("监听事件触发，event内容为："   event);
			}
		};
		setDataAsyncWithCallback(zkClient, callback, "/testcurator", "testing".getBytes());

		System.out.println(zkClient.getState());
		CloseableUtils.closeQuietly(zkClient);
	}

	// 创建默认节点
	public static void create(CuratorFramework client, String path, byte[] payload) throws Exception {
		// this will create the given ZNode with the given data
		client.create().forPath(path, payload);
	}

	// 创建临时节点
	public static void createEphemeral(CuratorFramework client, String path, byte[] payload) throws Exception {
		// this will create the given EPHEMERAL ZNode with the given data
		client.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload);
	}

	// 创建序列临时节点
	public static String createEphemeralSequential(CuratorFramework client, String path, byte[] payload)
			throws Exception {
		// this will create the given EPHEMERAL-SEQUENTIAL ZNode with the given data
		// using Curator protection.
		return client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, payload);
	}

	// 创建幂等的节点
	public static void createIdempotent(CuratorFramework client, String path, byte[] payload) throws Exception {
		/*
		 * This will create the given ZNode with the given data idempotently, meaning
		 * that if the initial create failed transiently, it will be retried and behave
		 * as if the first create never happened, even if the first create actually
		 * succeeded on the server but the client didn't know it.
		 */
		client.create().idempotent().forPath(path, payload);
	}

	// 针对节点，修改其数据
	public static void setData(CuratorFramework client, String path, byte[] payload) throws Exception {
		// set data for the given node
		client.setData().forPath(path, payload);
	}

	// 此监听主要针对background通知和错误通知。使用此监听器之后，调用inBackground方法会异步获得监听，而对于节点的创建或修改则不会触发监听事件
	public static void setDataAsync(CuratorFramework client, String path, byte[] payload) throws Exception {
		// this is one method of getting event/async notifications
		CuratorListener listener = new CuratorListener() {
			@Override
			public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
				// examine event for details
				System.out.println("监听事件触发，event内容为："   event);
			}
		};
		client.getCuratorListenable().addListener(listener);

		// 异步获取节点数据
		client.getData().inBackground().forPath(path);

		// 变更节点内容
		// set data for the given node asynchronously. The completion notification
		// is done via the CuratorListener.
		client.setData().inBackground().forPath(path, payload);
		Thread.sleep(Integer.MAX_VALUE);
	}

	// 通过回调函数获取数据变化
	public static void setDataAsyncWithCallback(CuratorFramework client, BackgroundCallback callback, String path,
			byte[] payload) throws Exception {
		// this is another method of getting notification of an async completion
		client.setData().inBackground(callback).forPath(path, payload);

	}

	//
	public static void setDataIdempotent(CuratorFramework client, String path, byte[] payload, int currentVersion)
			throws Exception {
		/*
		 * This will set the given ZNode with the given data idempotently, meaning that
		 * if the initial setData failed transiently, it will be retried and behave as
		 * if the first setData never happened, even if the first setData actually
		 * succeeded on the server but the client didn't know it. In other words, if
		 * currentVersion == X and payload = P, this will return success if the znode
		 * ends up in the state (version == X 1 && data == P). If withVersion is not
		 * specified, it will end up with success so long as the data == P, no matter
		 * the znode version.
		 */
		client.setData().idempotent().withVersion(currentVersion).forPath(path, payload);
		client.setData().idempotent().forPath(path, payload);
	}

	// 删除节点
	public static void delete(CuratorFramework client, String path) throws Exception {
		// delete the given node
		client.delete().forPath(path);
	}

	//
	public static void guaranteedDelete(CuratorFramework client, String path) throws Exception {
		// delete the given node and guarantee that it completes

		/*
		 * Guaranteed Delete
		 * 
		 * Solves this edge case: deleting a node can fail due to connection issues.
		 * Further, if the node was ephemeral, the node will not get auto-deleted as the
		 * session is still valid. This can wreak havoc with lock implementations.
		 * 
		 * 
		 * When guaranteed is set, Curator will record failed node deletions and attempt
		 * to delete them in the background until successful. NOTE: you will still get
		 * an exception when the deletion fails. But, you can be assured that as long as
		 * the CuratorFramework instance is open attempts will be made to delete the
		 * node.
		 */

		client.delete().guaranteed().forPath(path);
	}

	//
	public static void deleteIdempotent(CuratorFramework client, String path, int currentVersion) throws Exception {
		/*
		 * This will delete the given ZNode with the given data idempotently, meaning
		 * that if the initial delete failed transiently, it will be retried and behave
		 * as if the first delete never happened, even if the first delete actually
		 * succeeded on the server but the client didn't know it. In other words, if
		 * currentVersion == X, this will return success if the znode ends up deleted,
		 * and will retry after connection loss if the version the znode's version is
		 * still X. If withVersion is not specified, it will end up successful so long
		 * as the node is deleted eventually. Kind of like guaranteed but not in the
		 * background. For deletes this is equivalent to the older quietly() behavior,
		 * but it is also provided under idempotent() for compatibility with
		 * Create/SetData.
		 */
		client.delete().idempotent().withVersion(currentVersion).forPath(path);
		client.delete().idempotent().forPath(path);
		client.delete().quietly().withVersion(currentVersion).forPath(path);
		client.delete().quietly().forPath(path);
	}

	//
	public static List<String> watchedGetChildren(CuratorFramework client, String path) throws Exception {
		/**
		 * Get children and set a watcher on the node. The watcher notification will
		 * come through the CuratorListener (see setDataAsync() above).
		 */
		return client.getChildren().watched().forPath(path);
	}

	//
	public static List<String> watchedGetChildren(CuratorFramework client, String path, Watcher watcher)
			throws Exception {
		/**
		 * Get children and set the given watcher on the node.
		 */
		return client.getChildren().usingWatcher(watcher).forPath(path);
	}

	public static boolean isExists(CuratorFramework zkClient, String path) throws Exception {
		/*
		 * boolean flag = false; Stat stat = zkClient.checkExists().forPath(path); if
		 * (stat != null) { flag = true; }
		 */
		return zkClient.checkExists().forPath(path) != null ? true : false;
	}
}

6、监听示例

包含原生API和CuratorCache、CuratorListener
atomic，分布式计数器(DistributedAtomicLong)，能在分布式环境下实现原子自增
barriers，分布式屏障(DistributedBarrier)，使用屏障来阻塞分布式环境中进程的运行，直到满足特定的条件
cache，监听机制，分为NodeCache(监听节点数据变化)，PathChildrenCache（监听节点的子节点数据变化），TreeCache（既能监听自身节点数据变化也能监听子节点数据变化）
leader，leader选举
locks，分布式锁
nodes，提供持久化节点(PersistentNode)服务，即使客户端与zk服务的连接或者会话断开
queue，分布式队列(包括优先级队列DistributedPriorityQueue，延迟队列DistributedDelayQueue等)
shared，分布式计数器SharedCount
要使用该部分功能，则需要在pom.xml中增加如下部分

<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-recipes -->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.3.0</version>
</dependency>

代码示例如下：

import java.util.function.Consumer;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.CuratorCacheListenerBuilder.ChangeListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.zookeeper_curator.App;
import org.zookeeper_curator.Constant;

/**
 * @author chenw
 * TreeCacheListener、PathChildrenCacheListener和NodeCacheListener均已过期，替代的是CuratorCache
 */
public class ListenerDemo {
	// 创建连接实例
	private static CuratorFramework zkClient = null;

	public static void main(String[] args) throws Exception {
		zkClient = App.createWithOptions(Constant.zkServerAddress, Constant.retryPolicy, Constant.CONNECTION_TIMEOUT,
				Constant.SESSION_TIMEOUT);
		zkClient.start();

		String path = "/testNode";

		//
		if (!isExists(zkClient, path)) {
			zkClient.create().forPath(path, "testing".getBytes());
		}

//		watchEvent(zkClient, path);

//		curatorListener(zkClient, path);

//		curatorNodeCache(zkClient, path);

		curatorNodeCacheByLambda(zkClient, path);

		CloseableUtils.closeQuietly(zkClient);
	}

	// 使用原生zookeeper api，监控路径变化，只能监听一次变化
//	运行结果如下：
//	监听节点内容：testing
//	监听器watchedEvent：WatchedEvent state:SyncConnected type:NodeDataChanged path:/testNode
//实际数据如下：
//	[zk: server1:2118(CONNECTED) 39] get /testNode 
//	second data	
	private static void watchEvent(CuratorFramework zkClient, String path) throws Exception {

		byte[] data = zkClient.getData().usingWatcher(new Watcher() {
			@Override
			public void process(WatchedEvent watchedEvent) {
				System.out.println("监听器watchedEvent："   watchedEvent);
			}
		}).forPath(path);
		System.out.println("监听节点内容："   new String(data));

		// 第一次变更节点数据
		zkClient.setData().forPath(path, "new data".getBytes());

		// 第二次变更节点数据
		zkClient.setData().forPath(path, "second data".getBytes());
		Thread.sleep(Integer.MAX_VALUE);
	}

	// CuratorListener监听，此监听主要针对background通知和错误通知。使用此监听器之后，调用inBackground方法会异步获得监听，而对于节点的创建或修改则不会触发监听事件
	// 其中两次触发监听事件，第一次触发为注册监听事件时触发，第二次为getData异步处理返回结果时触发。而setData的方法并未触发监听事件。
	// 测试结果是只有getData数据时有结果返回

//	运行结果如下：
//	监听事件触发，event内容为：testing   CuratorEventImpl{type=GET_DATA, resultCode=0, path='/testNode', name='null', children=null, context=null, stat=55834578087,55834578087,1659404526966,1659404526966,0,0,0,0,7,0,55834578087
//			, data=[116, 101, 115, 116, 105, 110, 103], watchedEvent=null, aclList=null, opResults=null}

	// 实际结果
//	[zk: server1:2118(CONNECTED) 56] get /testNode 
//	testing123
	private static void curatorListener(CuratorFramework zkClient, String path) throws Exception {

		zkClient.getCuratorListenable().addListener(new CuratorListener() {
			// 增加监听方法
			public void eventReceived(CuratorFramework client, CuratorEvent event) {
				System.out.println("监听事件触发，event内容为："   new String(event.getData())   "   "   event);

			}
		});

		// 异步获取节点数据
		zkClient.getData().inBackground().forPath(path);

		// 变更节点内容
		zkClient.setData().forPath(path, "testing123".getBytes());

		Thread.sleep(Integer.MAX_VALUE);
	}

	// 该版本适合zookeeper 3.6.0 以上版本，服务器 zookeeper 服务也需要为 3.6.0 版本
	// CuratorCache会试图将来自节点的数据保存在本地缓存中。
	// 可以缓存指定的单个节点，也可以缓存以指定节点为根的整个子树（默认缓存方案）。可以给CuratorCache实例注册监听器，当相关节点发生更改时会接收到通知，
	// 将响应更新、创建、删除等事件。

	// 客户端操作结果
//	[zk: server2:2118(CONNECTED) 138] delete /testNode 
//	[zk: server2:2118(CONNECTED) 140] ls /
//	[zookeeper]
//	[zk: server2:2118(CONNECTED) 141] create /testNode
//	Created /testNode
//	[zk: server2:2118(CONNECTED) 142] ls /
//	[testNode, zookeeper]
//	[zk: server2:2118(CONNECTED) 143] set /testNode "testing1234"
//	[zk: server2:2118(CONNECTED) 144] get /testNode 
//	testing1234

//	程序运行结果
//	   create:ChildData{path='/testNode', stat=55834578106,55834578106,1659419377672,1659419377672,0,0,0,0,7,0,55834578106, data=[116, 101, 115, 116, 105, 110, 103]}
//	   初始化...
//	   delete:ChildData{path='/testNode', stat=55834578106,55834578106,1659419377672,1659419377672,0,0,0,0,7,0,55834578106, data=[116, 101, 115, 116, 105, 110, 103]}
//	   create:ChildData{path='/testNode', stat=55834578109,55834578109,1659419426777,1659419426777,0,0,0,0,0,0,55834578109, data=null}
//	   ChildData{path='/testNode', stat=55834578109,55834578109,1659419426777,1659419426777,0,0,0,0,0,0,55834578109 , data=null}   change  to ChildData{path='/testNode', stat=55834578109,55834578110,1659419426777,1659419448002,1,0,0,0,11,0,55834578109, data=[116, 101, 115, 116, 105, 110, 103, 49, 50, 51, 52]}
	private static void curatorNodeCache(CuratorFramework zkClient, String path) throws Exception {
		CuratorCache cache = CuratorCache.build(zkClient, path);

		CuratorCacheListener listener = CuratorCacheListener.builder().forCreates(new Consumer<ChildData>() {// 创建监控

			@Override
			public void accept(ChildData t) {
				System.out.println("   create:"   t);
			}
		}).forChanges(new ChangeListener() {// 改变监控

			@Override
			public void event(ChildData oldNode, ChildData node) {
				System.out.println(oldNode   "   change  to "   node);
			}
		}).forDeletes(new Consumer<ChildData>() {// 删除监控

			@Override
			public void accept(ChildData t) {
				System.out.println("   delete:"   t);
			}
		}).forInitialized(new Runnable() {// 初始化监控

			@Override
			public void run() {
				System.out.println("   初始化...");
			}
		}).build();
		// register the listener
		cache.listenable().addListener(listener);

		// the cache must be started
		cache.start();

		Thread.sleep(Integer.MAX_VALUE);
	}

	// 该方法与curatorNodeCache方法功能完全一致，不同的是写法
	private static void curatorNodeCacheByLambda(CuratorFramework zkClient, String path) throws Exception {

		CuratorCache cache = CuratorCache.build(zkClient, path);

		CuratorCacheListener listener = CuratorCacheListener.builder()
				.forCreates(node -> System.out.println("   create:"   node))
				.forChanges((oldNode, node) -> System.out.println(oldNode   "   change  to "   node))
				.forDeletes(node -> System.out.println("   delete:"   node))
				.forInitialized(() -> System.out.println("   初始化...")).build();

		// register the listener
		cache.listenable().addListener(listener);

		// the cache must be started
		cache.start();

		Thread.sleep(Integer.MAX_VALUE);
	}

	private static boolean isExists(CuratorFramework zkClient, String path) throws Exception {
		/*
		 * boolean flag = false; Stat stat = zkClient.checkExists().forPath(path); if
		 * (stat != null) { flag = true; }
		 */
		return zkClient.checkExists().forPath(path) != null ? true : false;
	}
}

7、计数器示例

1）、单机原子自增性实现

原子性就是指该操作是不可再分的，要么全部执行，要么全部不执行。
Java中保证原子性操作的有：

• Synchronized和Lock，加锁使得同一时刻只有一个线程能访问共享变量，操作自然是原子的
• java.util.concurrent.atomic下的原子操作类，如AtomicInteger，AtomicReference，基于Cas算法实现了类似乐观锁版本更新控制的原子操作

这两种方法，下面会分别介绍如何使用来解决i 原子性问题
volatile关键字不能保证原子性，保证原子性还得通过synchronized，Lock和java.util.concurrent.atomic下的原子操作类。

1、Synchronized示例

    static int count = 0;
    static CountDownLatch countDownLatch = new CountDownLatch(10);
    static Object lock = new Object();
    public static void main(String[] args) {
        //创建10个线程 每个线程内部自增100000次
        for (int i = 0; i < 10; i  ) {
            new Thread(() -> {
                for (int j = 0; j < 100000; j  ) {
	                //锁住lock对象，lock对象所有线程共享
                    synchronized (lock) {
                        count  ;
                    }
                }
                countDownLatch.countDown();
            }).start();
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(count);
    }

2、Lock示例

    static int count = 0;
    static CountDownLatch countDownLatch = new CountDownLatch(10);
    static ReentrantLock lock = new ReentrantLock();

    public static void main(String[] args) {
        //创建10个线程 每个线程内部自增100000次
        for (int i = 0; i < 10; i  ) {
            new Thread(() -> {
                for (int j = 0; j < 100000; j  ) {
                    lock.lock();
                    count  ;
                    lock.unlock();
                }
                countDownLatch.countDown();
            }).start();
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(count);
    }

ReentrantLock相比Synchronized，主要区别有以下几点

3、AtomicInteger示例

将count变量设为AtomicInteger，使用incrementAndGet方法也能实现，变量原子自增

    static CountDownLatch countDownLatch = new CountDownLatch(10);
    static AtomicInteger count = new AtomicInteger(0);
    public static void main(String[] args) {
        //创建10个线程 每个线程内部自增100000次
        for (int i = 0; i < 10; i  ) {
            new Thread(() -> {
                for (int j = 0; j < 100000; j  ) {
                    count.incrementAndGet();
                }
                countDownLatch.countDown();
            }).start();
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(count.get());
    }

2）、分布式线程安全原子自增实现

Curator基于Zookeeper实现的分布式计数器，Curator recipes包下实现了DistributedAtomicInteger，DistributedAtomicLong等分布式原子自增计数器。

import java.util.concurrent.CountDownLatch;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.zookeeper_curator.App;
import org.zookeeper_curator.Constant;

/**
 * @author alanchan
 *
 */
public class DistributedAtomicIntegerDemo {
	static CountDownLatch countDownLatch = new CountDownLatch(10);
	static int count = 0;
	// 创建连接实例
	private static CuratorFramework zkClient = null;

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		zkClient = App.createWithOptions(Constant.zkServerAddress, Constant.retryPolicy, Constant.CONNECTION_TIMEOUT,
				Constant.SESSION_TIMEOUT);
		zkClient.start();

		String path = "/countdown";
//		testIDoublePlus();
		distributedAtomicInteger(zkClient, path);
		CloseableUtils.closeQuietly(zkClient);
	}

	private static void distributedAtomicInteger(CuratorFramework zkClient, String path) throws Exception {
		DistributedAtomicInteger distributedAtomicInteger = new DistributedAtomicInteger(zkClient, path,
				new ExponentialBackoffRetry(1000, 3));
//		1.start
//        start作用是启动一个新线程。
//		当用start()开始一个线程后，线程就进入就绪状态，使线程所代表的虚拟处理机处于可运行状态，这意味着它可以由JVM调度并执行。但是这并不意味着线程就会立即运行。只有当cpu分配时间片时，这个线程获得时间片时，才开始执行run()方法。start()不能被重复调用，它调用run()方法.run()方法是你必须重写的
//		2.run
//		run()就和普通的成员方法一样，可以被重复调用。
//		如果直接调用run方法，并不会启动新线程！程序中依然只有主线程这一个线程，其程序执行路径还是只有一条，还是要顺序执行，还是要等待run方法体执行完毕后才可继续执行下面的代码，这样就没有达到多线程的目的。
//		调用start方法方可启动线程，而run方法只是thread的一个普通方法调用，还是在主线程里执行。
//下面的例子运行结果是10×100，具体看start和run的解释
		//		for (int i = 0; i < 10; i  ) {
//
//			Runnable thread = new Runnable() {
//				@Override
//				public void run() {
//					for (int j = 0; j < 100; j  ) {
//						try {
//							AtomicValue<Integer> result = distributedAtomicInteger.increment();
//						} catch (Exception e) {
//							e.printStackTrace();
//						}
//					}
//					countDownLatch.countDown();
//				}
//			};
//			thread.run();
//		}

		for (int i = 0; i < 10; i  ) {
			new Thread(() -> {
				for (int j = 0; j < 100; j  ) {
					try {
						// 调用add方法自增
						// Add delta to the current value and return the new value information. Remember
						// to always check {@link AtomicValue#succeeded()}.
						AtomicValue<Integer> result = distributedAtomicInteger.add(1);
//						if (result.succeeded()) {
//							System.out.println("current value: "   distributedAtomicInteger.get().postValue());
//						}
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				countDownLatch.countDown();

			}).start();
		}
		try {
			countDownLatch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		// 查看结果
		System.out.println("多线程自增结果:"   distributedAtomicInteger.get().postValue());
	}

	/**
	 * 测试i   是否线程安全
	 * CountDownLatch调用await()方法的线程将一直阻塞等待，直到这个CountDownLatch对象的计数值减到0(每次调用countDown方法计数器减一)为止。
	 * 例子里每个子线程自增100000次后调用countDown()方法将计数器减一，初始化数值10，10个线程全部跑完自增后，主线程await方法不再阻塞，输出count值
	 */
	private static void testIDoublePlus() {
		// 创建10个线程 每个线程内部自增100000次
		for (int i = 0; i < 10; i  ) {
			new Thread(() -> {
				for (int j = 0; j < 100000; j  ) {
					count  ;
				}
				countDownLatch.countDown();
			}).start();
		}
		try {
			countDownLatch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println(count);
	}

}

8、分布式锁示例

1）、实现原理

基于zookeeper瞬时有序节点实现的分布式锁。大致思想即为：每个客户端对某个功能加锁时，在zookeeper上的与该功能对应的指定节点的目录下，生成一个唯一的瞬时有序节点。判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。

2）、优点

锁安全性高，zk可持久化

3）、缺点

性能开销比较高。因为其需要动态产生、销毁瞬时节点来实现锁功能。

4）、实现

可以直接采用zookeeper第三方库curator即可方便地实现分布式锁。
1、使用原生API存在的问题

• 会话连接是异步的，需要自己去处理。比如使用 CountDownLatch Watch
• 需要重复注册，不然就不能生效
• 开发的复杂性比较高
• 不支持多节点删除和创建。需要自己去递归

2、Curator主要实现了下面四种锁

• InterProcessMutex：分布式可重入排它锁
• InterProcessSemaphoreMutex：分布式排它锁
• InterProcessReadWriteLock：分布式读写锁
• InterProcessMultiLock：将多个锁作为单个实体管理的容器

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.zookeeper_curator.App;
import org.zookeeper_curator.Constant;

/**
 * @author alanchan 测试分布式锁
 */
public class DistributedLockDemo {

	private static Integer testData = 0;
	private static final ExecutorService EXECUTOR_SERVICE = Executors.newCachedThreadPool();

	public static void main(String[] args) throws Exception {
		String path = "/locks";
  #InterProcessMutex：分布式可重入排它锁
//		testInterProcessMutex(path);
#InterProcessSemaphoreMutex：分布式排它锁
//		testInterProcessSemaphoreMutex(path);  //该方法不能获得期望的结果
#InterProcessReadWriteLock：分布式读写锁
//		pool-1-thread-4获取读锁
//		pool-1-thread-4释放读锁
//		pool-1-thread-2获取读锁
//		pool-1-thread-2释放读锁
//		pool-1-thread-5获取写锁
//		pool-1-thread-5释放写锁
//		pool-1-thread-3获取读锁
//		pool-1-thread-1获取读锁
//		pool-1-thread-3释放读锁
//		pool-1-thread-1释放读锁
		// 验证读写锁
        for (int i = 0; i < 5; i  ) {
            EXECUTOR_SERVICE.execute(new InterProcessReadWriteLockRunnable());
        }

//		pool-1-thread-5获取写锁
//		pool-1-thread-5获取读锁，锁降级成功
//		pool-1-thread-5释放读锁
//		pool-1-thread-5释放写锁
//		pool-1-thread-4获取写锁
//		pool-1-thread-4获取读锁，锁降级成功
//		pool-1-thread-4释放读锁
//		pool-1-thread-4释放写锁
//		pool-1-thread-2获取写锁
//		pool-1-thread-2获取读锁，锁降级成功
//		pool-1-thread-2释放读锁
//		pool-1-thread-2释放写锁
//		pool-1-thread-1获取写锁
//		pool-1-thread-1获取读锁，锁降级成功
//		pool-1-thread-1释放读锁
//		pool-1-thread-1释放写锁
//		pool-1-thread-3获取写锁
//		pool-1-thread-3获取读锁，锁降级成功
//		pool-1-thread-3释放读锁
//		pool-1-thread-3释放写锁
		// 验证锁降级
//		for (int i = 0; i < 5; i  ) {
//			EXECUTOR_SERVICE.execute(new InterProcessReadWriteLockRunnable1());
//		}
	}

	// 写入数据
	private static void write(String path) throws Exception {
		// 创建读写锁对象, Curator 以公平锁的方式进行实现
		InterProcessReadWriteLock lock = new InterProcessReadWriteLock(getCuratorFramework(), path);
		// 获取写锁(使用 InterProcessMutex 实现, 所以是可以重入的)
		InterProcessLock writeLock = lock.writeLock();
		writeLock.acquire();
		try {
			Thread.sleep(10);
			testData  ;
			System.out.println("写入数据，测试："   testData);
		} finally {
			writeLock.release();
		}
	}

	// 读取数据
	private void read(CuratorFramework zkClient, String path) throws Exception {
		// 创建读写锁对象, Curator 以公平锁的方式进行实现
		InterProcessReadWriteLock lock = new InterProcessReadWriteLock(zkClient, path);
		// 获取读锁(使用 InterProcessMutex 实现, 所以是可以重入的)
		InterProcessLock readLock = lock.readLock();
		readLock.acquire();
		try {
			Thread.sleep(10);
			System.out.println("读取数据，测试："   testData);
		} finally {
			readLock.release();
		}
	}

	// Result:
//	线程2 获取到锁
//	线程2 再次获取到锁
//	线程2 释放锁
//	线程2  再次释放锁
//	线程1 获取到锁
//	线程1 再次获取到锁
//	线程1 释放锁
//	线程1  再次释放锁
	// 共享可重入锁
	public static void testInterProcessMutex(String path) {
		// 创建分布式锁1
		InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), path);
		// 创建分布式锁2
		InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), path);
		new Thread(new Runnable() {

			@Override
			public void run() {

				try {
					lock1.acquire();
					System.out.println("线程1 获取到锁");
					lock1.acquire();
					System.out.println("线程1 再次获取到锁");

					Thread.sleep(5 * 1000);

					lock1.release();
					System.out.println("线程1 释放锁");
					Thread.sleep(5 * 1000);
					lock1.release();
					System.out.println("线程1 再次释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();

		new Thread(new Runnable() {
			@Override
			public void run() {

				try {
					lock2.acquire();
					System.out.println("线程2 获取到锁");
					lock2.acquire();
					System.out.println("线程2 再次获取到锁");

					Thread.sleep(5 * 1000);

					lock2.release();
					System.out.println("线程2 释放锁");
					Thread.sleep(5 * 1000);
					lock2.release();
					System.out.println("线程2 再次释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
	}

	// result:
	// 线程2 获取到锁
	// 然后处于阻塞状态，其他线程不能获取到锁，也不能释放锁

	// 共享不可重入锁
	public static void testInterProcessSemaphoreMutex(String path) {
		// 创建分布式锁1
		InterProcessLock lock1 = new InterProcessSemaphoreMutex(getCuratorFramework(), path);
		// 创建分布式锁2
		InterProcessLock lock2 = new InterProcessSemaphoreMutex(getCuratorFramework(), path);
		new Thread(new Runnable() {

			@Override
			public void run() {

				try {
					lock1.acquire();
					System.out.println("线程1 获取到锁");
					lock1.acquire();
					System.out.println("线程1 再次获取到锁");

					Thread.sleep(5 * 1000);

					lock1.release();
					System.out.println("线程1 释放锁");
					Thread.sleep(5 * 1000);
					lock1.release();
					System.out.println("线程1 再次释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();

		new Thread(new Runnable() {
			@Override
			public void run() {

				try {
					lock2.acquire();
					System.out.println("线程2 获取到锁");
					lock2.acquire();
					System.out.println("线程2 再次获取到锁");

					Thread.sleep(5 * 1000);

					lock2.release();
					System.out.println("线程2 释放锁");
					Thread.sleep(5 * 1000);
					lock2.release();
					System.out.println("线程2 再次释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
	}

	// 共享不可重入锁
	public InterProcessLock getInterProcessSemaphoreMutex(CuratorFramework zkClient, String path) {
		InterProcessLock ipsmLock = new InterProcessSemaphoreMutex(zkClient, path);
		return ipsmLock;
	}

	// 共享可重入锁
	public InterProcessLock getInterProcessMutex(CuratorFramework zkClient, String path) {
		InterProcessLock ipLock = new InterProcessMutex(zkClient, path);
		return ipLock;
	}

	// 共享可重入读锁
	public InterProcessLock getInterProcessReadLock(CuratorFramework zkClient, String path) {
		InterProcessReadWriteLock ipReadWriteLock = new InterProcessReadWriteLock(zkClient, path);
		InterProcessLock readLock = ipReadWriteLock.readLock();
		return readLock;
	}

	// 共享可重入写锁
	public InterProcessLock getInterProcessWriteLock(CuratorFramework zkClient, String path) {
		InterProcessReadWriteLock ipReadWriteLock = new InterProcessReadWriteLock(zkClient, path);
		InterProcessLock writeLock = ipReadWriteLock.writeLock();
		return writeLock;
	}

	private static CuratorFramework getCuratorFramework() {
		CuratorFramework client = App.createWithOptions(Constant.zkServerAddress, Constant.retryPolicy,
				Constant.CONNECTION_TIMEOUT, Constant.SESSION_TIMEOUT);
		// 启动客户端
		client.start();
		System.out.println("zookeeper 启动成功");
		return client;
	}
}

读写锁

import java.nio.charset.StandardCharsets;
import java.util.Random;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.zookeeper_curator.App;
import org.zookeeper_curator.Constant;

/**
 * @author alanchan
 * 测试读写锁
 */
public class InterProcessReadWriteLockRunnable implements Runnable {

	@Override
	public void run() {
		CuratorFramework zkClient = getCuratorFramework();
		String path = "/locks";
		// 创建InterProcessReadWriteLock实例，用于提供分布式锁的功能
		InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(zkClient, path,
				"分布式读写锁".getBytes(StandardCharsets.UTF_8));

		// 根据随机数来决定获取写锁还是读锁
		Random random = new Random();
		try {
			if (random.nextInt(10000) > 5000) {
				// 获取写锁
				readWriteLock.writeLock().acquire();
				System.out.println(Thread.currentThread().getName()   "获取写锁");
				Thread.sleep(2);
				System.out.println(Thread.currentThread().getName()   "释放写锁");
				// 释放写锁
				readWriteLock.writeLock().release();
			} else {
				// 获取读锁
				readWriteLock.readLock().acquire();
				System.out.println(Thread.currentThread().getName()   "获取读锁");
				Thread.sleep(2);
				System.out.println(Thread.currentThread().getName()   "释放读锁");
				// 释放读锁
				readWriteLock.readLock().release();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	private static CuratorFramework getCuratorFramework() {
		CuratorFramework client = App.createWithOptions(Constant.zkServerAddress, Constant.retryPolicy,
				Constant.CONNECTION_TIMEOUT, Constant.SESSION_TIMEOUT);
		// 启动客户端
		client.start();
		System.out.println("zookeeper 启动成功");
		return client;
	}
}

锁降级

import java.nio.charset.StandardCharsets;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.zookeeper_curator.App;
import org.zookeeper_curator.Constant;

/**
 * @author alanchan
 * 锁降级
 */
public class InterProcessReadWriteLockRunnable1 implements Runnable {

	private static CuratorFramework getCuratorFramework() {
		CuratorFramework client = App.createWithOptions(Constant.zkServerAddress, Constant.retryPolicy,
				Constant.CONNECTION_TIMEOUT, Constant.SESSION_TIMEOUT);
		// 启动客户端
		client.start();
		System.out.println("zookeeper 启动成功");
		return client;
	}

	@Override
	public void run() {
		CuratorFramework zkClient = getCuratorFramework();
		String path = "/locks";
		// 创建InterProcessReadWriteLock实例，用于提供分布式锁的功能
		InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(zkClient, path,
				"分布式读写锁".getBytes(StandardCharsets.UTF_8));
		try {
			// 获取写锁
			readWriteLock.writeLock().acquire();
			System.out.println(Thread.currentThread().getName()   "获取写锁");
			Thread.sleep(2000);
			// 锁降级
			readWriteLock.readLock().acquire();
			System.out.println(Thread.currentThread().getName()   "获取读锁，锁降级成功");
			Thread.sleep(2000);
			// 释放读锁
			System.out.println(Thread.currentThread().getName()   "释放读锁");
			readWriteLock.readLock().release();
			// 释放写锁
			System.out.println(Thread.currentThread().getName()   "释放写锁");
			readWriteLock.writeLock().release();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

以上简单的介绍了一下zookeeper的java三种客户端介绍-Curator（crud、事务操作、监听、分布式计数器、分布式锁）。具体深入的使用可能需要进一步的了解。

这篇好文章是转载于：学新通技术网