Zookeeper（八）开源客户端Curator的高级属性
2017-02-20 22:14
288 查看
到目前为止，我们已经知道如何使用Curator来创建会话，创建节点，删除节点，读取数据和更新数据等操作。值得一提的是，在前面都使用了Curator框架提供的同步接口，现在我们讲解如何通过Curator实现异步操作。
Curator中引入了BackgroundCallback接口，用来处理异步接口调用之后服务端返回的结果信息，其接口定义如下。
清单5-31
BackgroundCallback接口只有一个processResult方法，从注释中可以看出，该方法会在操作完成后被异步调用。该方法的参数说明如表5-23所示：
对于BackgroundCallback接口，我们重点来看CuratorEvent这个参数。CuratorEvent定义了ZooKeeper服务端发送到客户端的一系列事件参数，其中比较重要的有事件类型和响应码两个参数。
下面，我们通过一个实际例子来看看如何使用Curator异步接口。
运行程序，输出结果如下：
Main thread：main
event[code:0,type:CREATE]
Thread of prodessResutl:pool-3-thread-1
event[code:-110,type:CREATE]
Thread of prodessResult:maijj-EventThread
上面这个程序使用了异步接口inBackground来创建节点，前后两次调用，创建的节点名相同。从两次返回的event可以看出，第一次返回的响应码是0，表明此次次调用成功，即创建节点成功；而第二次返回的响应码是-110，表明该节点已经存在，无法重复创建。这些响应码和ZooKeeper原生的响应码是一致的。
我们来关注一下executor这个参数。在ZooKeeeper中，所有的异步通知事件都是由EventThread这个线程来处理的——EventThread线程用于穿行处理所有的事件通知。EventThread的“串行处理机制”在绝大部分应用场景下能够保证对事件处理的顺序性，但这个特性也有其弊端，就是一旦碰上一个复杂的处理单元，就会消耗过程的处理时间，从而影响对其他事件得分处理。因此，在上面的inBacigorund接口中，允许用户传入一个Executor实例，这样一来，就可以把那些复杂的事件处理放到一个专门的线程池中去，如Exceutors.newFixedThreadPool(2)。
所以在我们上面前后两次调用inBackground接口时传入的Executor参数。第一次传入了一个ExcecutorService，这样一来，Curatro的异步事件处理就会交由线程池去做。而第二次调用时，没有传入任何Executor，因此会使用ZooKeeper默认的EventThread来处理。
public NodeCache(CuratorFramework client,String path);
public NodeCache(CuratorFramework client,Strinig path,boolean dataIsCompressed);
NodeCache构造方法参数说明如表5-24所示。
同时，NodeCache定义了事件处理的回调接口NodeCacheListener。
清单5-34 NodeChacheListener回调接口定义
public interface NodeCacheListener{
public voidnodeChanged() throws Exception;
}
当数据节点的内容发生变化的时候，就会回调该方法。下面通过一个实际例子来看看如何在代码中使用NodeCache。
清单5-35 NodeCache使用示例
在上面的示例程序中，首先构造了一个NodeCache实例，然后调用start方法，该方法有个boolean类型的参数，默认是false，如果设置为ture，那么NodeCache在第一次启动的时候就会立刻从ZooKeeper上读取对应节点的数据内容，并保存在Cache中。
NodeCache不仅可以用于监听数据节点的内容变更，也能监听指定节点是否存在。如果原节点不存在，那么Cache就会在节点被创建后触发NodeCacheListener，但是，如果该节点被删除，那么Curatro就无法触发NodeCacheLIsten了。
PathChildrenCache定义了事件处理的回调接口PathChildrenCacheListener，其定义如下。
清单5-36 PathChildrenCacheListener回调定义
public interface PathChildrenCacheListener{
public void childEvent(CuratorFramework client,PathChildrenCacheEvent event) throws Exception;
}
当指定节点的子节点发生变化时，就会回调该方法。PathChildrenCacheEvent类中定义了所有的事件类型，主要包括新增子节点（CHILD_ADDED），子节点数据变更（CHILD_UPDATE）和子节点删除（CHILD_REMOVED）三类。
下面通过一个实际例子来看看如何在代码中使用PathChildrenCache。
清单5-37
运行程序，输出结果如下：
CHILD_ADDED，/zk-book/c1
CHILD_REMOVED，/zk-book/c1
在上面这个示例程序中，对/zk-book节点进行了子节点变更事件的监听，一旦该节点新增/删除子节点，或者子节点数据发生变更，就会回调PathChildrenCacheListener，并根据对应的事件类型进行相关的处理。同时，我们也看到，对于节点/zk-book本身的变更，并没有通知到客户端。
另外，和其他ZooKeeper客户端产品一样，Curator也无法对二级子节点进行事件监听。也就是说，如果使用PathChildrenCache对/zk-book进行监听，那么当/zk-book/c1/c2节点被创建或删除的时候，是无法触发子节点变更事件的。
二、Recipes模块
主要有
Elections(选举)，Locks（锁），Barriers（关卡），Atomic（原子量），Caches，Queues等
1、 Elections
选举主要依赖于LeaderSelector和LeaderLatch2个类。前者是所有存活的客户端不间断的轮流做Leader，大同社会。后者是一旦选举出Leader，除非有客户端挂掉重新触发选举，否则不会交出领导权。某党？
这两者在实现上是可以切换的，直接上代码，怎么切换注释里有。由于篇幅所限，这里仅贴出基于LeaderSelector的选举，更多代码见附件
Java代码
/**
* 本类基于leaderSelector实现,所有存活的client会公平的轮流做leader
* 如果不想频繁的变化Leader，需要在takeLeadership方法里阻塞leader的变更！ 或者使用 {@link}
* LeaderLatchClient
*/
public class LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {
private final String name;
private final LeaderSelector leaderSelector;
private final String PATH = "/leaderselector";
public LeaderSelectorClient(CuratorFramework client, String name) {
this.name = name;
leaderSelector = new LeaderSelector(client, PATH, this);
leaderSelector.autoRequeue();
}
public void start() throws IOException {
leaderSelector.start();
}
@Override
public void close() throws IOException {
leaderSelector.close();
}
/**
* client成为leader后，会调用此方法
*/
@Override
public void takeLeadership(CuratorFramework client) throws Exception {
int waitSeconds = (int) (5 * Math.random()) + 1;
System.out.println(name + "是当前的leader");
try {
Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
System.out.println(name + " 让出领导权\n");
}
}
Java代码
/**
* leader选举
*
* @author shencl
*/
public class LeaderSelectorExample {
public static void main(String[] args) {
List<CuratorFramework> clients = Lists.newArrayList();
List<LeaderSelectorClient> examples = Lists.newArrayList();
try {
for (int i = 0; i < 10; i++) {
CuratorFramework client = ClientFactory.newClient();
LeaderSelectorClient example = new LeaderSelectorClient(client, "Client #" + i);
clients.add(client);
examples.add(example);
client.start();
example.start();
}
System.out.println("----------先观察一会选举的结果-----------");
Thread.sleep(10000);
System.out.println("----------关闭前5个客户端，再观察选举的结果-----------");
for (int i = 0; i < 5; i++) {
clients.get(i).close();
}
// 这里有个小技巧，让main程序一直监听控制台输入，异步的代码就可以一直在执行。不同于while(ture)的是，按回车或esc可退出
new BufferedReader(new InputStreamReader(System.in)).readLine();
} catch (Exception e) {
e.printStackTrace();
} finally {
for (LeaderSelectorClient exampleClient : examples) {
CloseableUtils.closeQuietly(exampleClient);
}
for (CuratorFramework client : clients) {
CloseableUtils.closeQuietly(client);
}
}
}
}
2、locks
curator lock相关的实现在recipes.locks包里。顶级接口都是InterProcessLock。我们直接看最有代表性的InterProcessReadWriteLock 进程内部读写锁（可重入读写锁）。什么叫可重入，什么叫读写锁。不清楚的先查好资料吧。总之读写锁一定是成对出现的。
简易传送门
我们先定义两个任务，可并行的执行的，和互斥执行的。
Java代码
/**
* 并行任务
*
* @author shencl
*/
public class ParallelJob implements Runnable {
private final String name;
private final InterProcessLock lock;
// 锁等待时间
private final int wait_time = 5;
ParallelJob(String name, InterProcessLock lock) {
this.name = name;
this.lock = lock;
}
@Override
public void run() {
try {
doWork();
} catch (Exception e) {
// ingore;
}
}
public void doWork() throws Exception {
try {
if (!lock.acquire(wait_time, TimeUnit.SECONDS)) {
System.err.println(name + "等待" + wait_time + "秒，仍未能获取到lock,准备放弃。");
}
// 模拟job执行时间0-4000毫秒
int exeTime = new Random().nextInt(4000);
System.out.println(name + "开始执行,预计执行时间= " + exeTime + "毫秒----------");
Thread.sleep(exeTime);
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.release();
}
}
}
Java代码
/**
* 互斥任务
*
* @author shencl
*/
public class MutexJob implements Runnable {
private final String name;
private final InterProcessLock lock;
// 锁等待时间
private final int wait_time = 10;
MutexJob(String name, InterProcessLock lock) {
this.name = name;
this.lock = lock;
}
@Override
public void run() {
try {
doWork();
} catch (Exception e) {
// ingore;
}
}
public void doWork() throws Exception {
try {
if (!lock.acquire(wait_time, TimeUnit.SECONDS)) {
System.err.println(name + "等待" + wait_time + "秒，仍未能获取到lock,准备放弃。");
}
// 模拟job执行时间0-2000毫秒
int exeTime = new Random().nextInt(2000);
System.out.println(name + "开始执行,预计执行时间= " + exeTime + "毫秒----------");
Thread.sleep(exeTime);
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.release();
}
}
}
锁测试代码
Java代码
/**
* 分布式锁实例
*
* @author shencl
*/
public class DistributedLockExample {
private static CuratorFramework client = ClientFactory.newClient();
private static final String PATH = "/locks";
// 进程内部（可重入）读写锁
private static final InterProcessReadWriteLock lock;
// 读锁
private static final InterProcessLock readLock;
// 写锁
private static final InterProcessLock writeLock;
static {
client.start();
lock = new InterProcessReadWriteLock(client, PATH);
readLock = lock.readLock();
writeLock = lock.writeLock();
}
public static void main(String[] args) {
try {
List<Thread> jobs = Lists.newArrayList();
for (int i = 0; i < 10; i++) {
Thread t = new Thread(new ParallelJob("Parallel任务" + i, readLock));
jobs.add(t);
}
for (int i = 0; i < 10; i++) {
Thread t = new Thread(new MutexJob("Mutex任务" + i, writeLock));
jobs.add(t);
}
for (Thread t : jobs) {
t.start();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
CloseableUtils.closeQuietly(client);
}
}
}
看到没，用法和java concurrent包里的ReentrantReadWriteLock 是一模一样的。
事实上，整个recipes包的目录结构、实现原理同java concurrent包的设置是很一致的。比如有queue，Semaphore，Barrier等类，。他整个就是模仿jdk的实现，只不过是基于分布式的！
后边的几项，Barriers（关卡），Atomic（原子量），Caches，Queues和java concurrent包里的类的用法是一样的，就不继续贴了，有些附件里有。
要说明的是：有的功能性能不是特别理想，网上也没见有大的项目的使用案例。比如基于CAS机制的atomic，在某些情况重试的效率还不如硬同步，要是zookeeper节点再一多，各个节点之间通过event触发的数据同步极其频繁。那性能可以想象。
三、测试方法
curator提供了很好的测试工具，你甚至是可以在完全没有搭建zookeeper server端的情况下，完成测试。
有2个重要的类
TestingServer 模拟单点， TestingCluster模拟集群。
需要使用的话，得依赖
Xml代码
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<version>2.5.0</version>
</dependency>
Curator中引入了BackgroundCallback接口，用来处理异步接口调用之后服务端返回的结果信息，其接口定义如下。
清单5-31
public interface BackgroundCallback{
publc void processResult(CuratorFramework client, CuratorEvent event) throws Exception;
}BackgroundCallback接口只有一个processResult方法，从注释中可以看出，该方法会在操作完成后被异步调用。该方法的参数说明如表5-23所示：
| 参数名 | 说明 |
| client | 当前客户端实例 |
| event | 服务端事件 |
下面，我们通过一个实际例子来看看如何使用Curator异步接口。
public class Craate_Node_Background implements Watcher{
static String path = "/zk-book";
static CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString("domain1.book.zookeeper:2181")
.sessionTimeoutMs(5000)
.retryPolicy(new ExponentialBackoffRetry(1000,3))
.build();
static CountDownLatch semaphore = new CountDownLatch(2);
static ExecutorService tp = Executors.newFixedThreadPool(2);
public static void main(String[] args) {
client.start();
System.out.println("Main thread:"+Thread.currentThread.getName());
//此处传入了自定义的Executor
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
.inBackground(new BackgroundCallback()){
@Override
public void processResult(CuratorFramework client,CuratorEvent event)
throws Exception{
System.out.println("event[code:"+event.getResutlCode()+",type:"+
event.getType()+"]");
System.out.println("Thread of processResult:"+Thread.currentThread().getName());
}
},tp).forPath(path,"init".getBytes());
//此处没有传入自定义的Executro
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)\
.inBackground(new BackgorundCallback){
@Override
public void processResult(CuratorFramework client,CuratorEvent event) throws Exception{
System.out.println("event[code:"+event.getResultCode()+",type:"
+event.getType()+"]");
System.out.println("Thread of processResult:"+Thread.currentThread().getName());
semaphore.countDown();
}
}).forPath(path,"init".getBytes());
semaphore.await();
tp.shutdown();
}
}运行程序，输出结果如下：
Main thread：main
event[code:0,type:CREATE]
Thread of prodessResutl:pool-3-thread-1
event[code:-110,type:CREATE]
Thread of prodessResult:maijj-EventThread
上面这个程序使用了异步接口inBackground来创建节点，前后两次调用，创建的节点名相同。从两次返回的event可以看出，第一次返回的响应码是0，表明此次次调用成功，即创建节点成功；而第二次返回的响应码是-110，表明该节点已经存在，无法重复创建。这些响应码和ZooKeeper原生的响应码是一致的。
我们来关注一下executor这个参数。在ZooKeeeper中，所有的异步通知事件都是由EventThread这个线程来处理的——EventThread线程用于穿行处理所有的事件通知。EventThread的“串行处理机制”在绝大部分应用场景下能够保证对事件处理的顺序性，但这个特性也有其弊端，就是一旦碰上一个复杂的处理单元，就会消耗过程的处理时间，从而影响对其他事件得分处理。因此，在上面的inBacigorund接口中，允许用户传入一个Executor实例，这样一来，就可以把那些复杂的事件处理放到一个专门的线程池中去，如Exceutors.newFixedThreadPool(2)。
所以在我们上面前后两次调用inBackground接口时传入的Executor参数。第一次传入了一个ExcecutorService，这样一来，Curatro的异步事件处理就会交由线程池去做。而第二次调用时，没有传入任何Executor，因此会使用ZooKeeper默认的EventThread来处理。
典型使用场景
Curator不仅为开发者提供了更为便利的API接口，而且还提供了一些典型场景的使用参考。读者可以这些使用参考中更好地理解如何使用ZooKeeper客户但。这些使用参考都在recipes包中，读者需要单独依赖一下Maven依赖来获取：<dependency> <groudId>org.apache.curator</groudId> <artifactId>curator-recipes</artifactId> <version>2.4.2</version> </dependency>
事件监听
ZooKeeper原生支持通过注册Watcher来进行事件监听，但是其使用并不方便需要开发人员反复注册Watcher，比较繁琐。Curator引入了Cache来实现对ZooKeeper事件的监听。Cache是Curator中对事件监听的包装，其对事件监听其实可以近似看作是i一个本地缓存视图和远程ZooKeeper视图的对比过程。同时Curator能够自动为开发人员处理反复注册监听，从而大大简化了原生API开发的繁琐过程。Cache分为两类监听类型：节点监听和子节点监听。NodeCache
NodeCache用于监听指定ZooKeeper数据节点本身的变化，其构造方法有如下两个：public NodeCache(CuratorFramework client,String path);
public NodeCache(CuratorFramework client,Strinig path,boolean dataIsCompressed);
NodeCache构造方法参数说明如表5-24所示。
| 参数名 | 说明 |
| client | Curatro客户端实例 |
| path | 数据节点的节点路径 |
| dataIsCompressed | 是否进行数据压缩 |
清单5-34 NodeChacheListener回调接口定义
public interface NodeCacheListener{
public voidnodeChanged() throws Exception;
}
当数据节点的内容发生变化的时候，就会回调该方法。下面通过一个实际例子来看看如何在代码中使用NodeCache。
清单5-35 NodeCache使用示例
public class NodeCache_Sample{
static String path = "/zk-book/nodecache";
static CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString("domain1.book.zookeeper:2181")
.sessionTimeoutMs(5000)
.retryPolicy(new ExponentialBackoffRetry(1000,3))
.build();
public static void main(String[] args) {
client.start();
client.create().creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL).forPath(path,"init".getBytes());
final NodeCache cache = new NodeCache(client,path,false);
cache.start(true);
cache.getListenable().addListener(new NodeCacheListener){
@Override
public void nodeChanged() throws Exception{
System.out.println("Node data update,new data:"+
new String(cache.getCurrentData().getData()));
}
});
client.setData().forPath(path,"u".getBytes());
Thread.sleep(1000);
client.delete().deletingChildrenIfNeeded().forPath(path);
Thread.sleep(Integer.MAX_VALUE);
}
}在上面的示例程序中，首先构造了一个NodeCache实例，然后调用start方法，该方法有个boolean类型的参数，默认是false，如果设置为ture，那么NodeCache在第一次启动的时候就会立刻从ZooKeeper上读取对应节点的数据内容，并保存在Cache中。
NodeCache不仅可以用于监听数据节点的内容变更，也能监听指定节点是否存在。如果原节点不存在，那么Cache就会在节点被创建后触发NodeCacheListener，但是，如果该节点被删除，那么Curatro就无法触发NodeCacheLIsten了。
PathChildrenCache
pathChildrenCache用于监听指定ZooKeeper数据节点的子节点变化情况。PathChildrenCache定义了事件处理的回调接口PathChildrenCacheListener，其定义如下。
清单5-36 PathChildrenCacheListener回调定义
public interface PathChildrenCacheListener{
public void childEvent(CuratorFramework client,PathChildrenCacheEvent event) throws Exception;
}
当指定节点的子节点发生变化时，就会回调该方法。PathChildrenCacheEvent类中定义了所有的事件类型，主要包括新增子节点（CHILD_ADDED），子节点数据变更（CHILD_UPDATE）和子节点删除（CHILD_REMOVED）三类。
下面通过一个实际例子来看看如何在代码中使用PathChildrenCache。
清单5-37
public class PathChildrenCache_Sample{
static String path = "/zk-book";
static CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString("domain1.book.zookeeper:2181")
.sessionTimeoutMs(5000)
.retryPolicy(new ExponentialBackoffRetry(1000,3))
.build();
public static void main(String[] args) {
client.start();
PathChildrenCache cache = new PathChildrenCache(client,path,true);
cache.start(StartMode.POST_INITIALIZED_EVENT);
cache.getListenabel().addListener(new PathChildrenCacheListener(){
public void childEvent(CuratorFramework client,PathChildrenCacheEvent event) throws Exceptiion{
switch(event.getType()){
case CHILD_ADDED:
System.out.println("CHILD_ADDED,"+event.getData().getPath());
break;
case CHILD_UPDATE:
System.out.println("CHILD_UPDATE,"+event.getDate().getPath());
break;
case CHILD_REMOVED:
System.out.println("CHILD_REMOVED,"+event.getData().getPath());
break;
default:
break;
}
}
});
client.create().withMode(CreateMode.PERSISTENT).froPath(path);
Thread.sleep(1000);
client.create().withMOde(CreateMode.PERSISTENT).forPath(path+"/c1");
Thread.sleep(1000);
client.delete().forPath(path+"/c1");
Thread.sleep(1000);
client.delete().forPath(path);
Thread.sleep(Integer.MAX_VALUE);
}
}运行程序，输出结果如下：
CHILD_ADDED，/zk-book/c1
CHILD_REMOVED，/zk-book/c1
在上面这个示例程序中，对/zk-book节点进行了子节点变更事件的监听，一旦该节点新增/删除子节点，或者子节点数据发生变更，就会回调PathChildrenCacheListener，并根据对应的事件类型进行相关的处理。同时，我们也看到，对于节点/zk-book本身的变更，并没有通知到客户端。
另外，和其他ZooKeeper客户端产品一样，Curator也无法对二级子节点进行事件监听。也就是说，如果使用PathChildrenCache对/zk-book进行监听，那么当/zk-book/c1/c2节点被创建或删除的时候，是无法触发子节点变更事件的。
二、Recipes模块
主要有
Elections(选举)，Locks（锁），Barriers（关卡），Atomic（原子量），Caches，Queues等
1、 Elections
选举主要依赖于LeaderSelector和LeaderLatch2个类。前者是所有存活的客户端不间断的轮流做Leader，大同社会。后者是一旦选举出Leader，除非有客户端挂掉重新触发选举，否则不会交出领导权。某党？
这两者在实现上是可以切换的，直接上代码，怎么切换注释里有。由于篇幅所限，这里仅贴出基于LeaderSelector的选举，更多代码见附件
Java代码
/**
* 本类基于leaderSelector实现,所有存活的client会公平的轮流做leader
* 如果不想频繁的变化Leader，需要在takeLeadership方法里阻塞leader的变更！ 或者使用 {@link}
* LeaderLatchClient
*/
public class LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {
private final String name;
private final LeaderSelector leaderSelector;
private final String PATH = "/leaderselector";
public LeaderSelectorClient(CuratorFramework client, String name) {
this.name = name;
leaderSelector = new LeaderSelector(client, PATH, this);
leaderSelector.autoRequeue();
}
public void start() throws IOException {
leaderSelector.start();
}
@Override
public void close() throws IOException {
leaderSelector.close();
}
/**
* client成为leader后，会调用此方法
*/
@Override
public void takeLeadership(CuratorFramework client) throws Exception {
int waitSeconds = (int) (5 * Math.random()) + 1;
System.out.println(name + "是当前的leader");
try {
Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
System.out.println(name + " 让出领导权\n");
}
}
Java代码
/**
* leader选举
*
* @author shencl
*/
public class LeaderSelectorExample {
public static void main(String[] args) {
List<CuratorFramework> clients = Lists.newArrayList();
List<LeaderSelectorClient> examples = Lists.newArrayList();
try {
for (int i = 0; i < 10; i++) {
CuratorFramework client = ClientFactory.newClient();
LeaderSelectorClient example = new LeaderSelectorClient(client, "Client #" + i);
clients.add(client);
examples.add(example);
client.start();
example.start();
}
System.out.println("----------先观察一会选举的结果-----------");
Thread.sleep(10000);
System.out.println("----------关闭前5个客户端，再观察选举的结果-----------");
for (int i = 0; i < 5; i++) {
clients.get(i).close();
}
// 这里有个小技巧，让main程序一直监听控制台输入，异步的代码就可以一直在执行。不同于while(ture)的是，按回车或esc可退出
new BufferedReader(new InputStreamReader(System.in)).readLine();
} catch (Exception e) {
e.printStackTrace();
} finally {
for (LeaderSelectorClient exampleClient : examples) {
CloseableUtils.closeQuietly(exampleClient);
}
for (CuratorFramework client : clients) {
CloseableUtils.closeQuietly(client);
}
}
}
}
2、locks
curator lock相关的实现在recipes.locks包里。顶级接口都是InterProcessLock。我们直接看最有代表性的InterProcessReadWriteLock 进程内部读写锁（可重入读写锁）。什么叫可重入，什么叫读写锁。不清楚的先查好资料吧。总之读写锁一定是成对出现的。
简易传送门
我们先定义两个任务，可并行的执行的，和互斥执行的。
Java代码
/**
* 并行任务
*
* @author shencl
*/
public class ParallelJob implements Runnable {
private final String name;
private final InterProcessLock lock;
// 锁等待时间
private final int wait_time = 5;
ParallelJob(String name, InterProcessLock lock) {
this.name = name;
this.lock = lock;
}
@Override
public void run() {
try {
doWork();
} catch (Exception e) {
// ingore;
}
}
public void doWork() throws Exception {
try {
if (!lock.acquire(wait_time, TimeUnit.SECONDS)) {
System.err.println(name + "等待" + wait_time + "秒，仍未能获取到lock,准备放弃。");
}
// 模拟job执行时间0-4000毫秒
int exeTime = new Random().nextInt(4000);
System.out.println(name + "开始执行,预计执行时间= " + exeTime + "毫秒----------");
Thread.sleep(exeTime);
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.release();
}
}
}
Java代码
/**
* 互斥任务
*
* @author shencl
*/
public class MutexJob implements Runnable {
private final String name;
private final InterProcessLock lock;
// 锁等待时间
private final int wait_time = 10;
MutexJob(String name, InterProcessLock lock) {
this.name = name;
this.lock = lock;
}
@Override
public void run() {
try {
doWork();
} catch (Exception e) {
// ingore;
}
}
public void doWork() throws Exception {
try {
if (!lock.acquire(wait_time, TimeUnit.SECONDS)) {
System.err.println(name + "等待" + wait_time + "秒，仍未能获取到lock,准备放弃。");
}
// 模拟job执行时间0-2000毫秒
int exeTime = new Random().nextInt(2000);
System.out.println(name + "开始执行,预计执行时间= " + exeTime + "毫秒----------");
Thread.sleep(exeTime);
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.release();
}
}
}
锁测试代码
Java代码
/**
* 分布式锁实例
*
* @author shencl
*/
public class DistributedLockExample {
private static CuratorFramework client = ClientFactory.newClient();
private static final String PATH = "/locks";
// 进程内部（可重入）读写锁
private static final InterProcessReadWriteLock lock;
// 读锁
private static final InterProcessLock readLock;
// 写锁
private static final InterProcessLock writeLock;
static {
client.start();
lock = new InterProcessReadWriteLock(client, PATH);
readLock = lock.readLock();
writeLock = lock.writeLock();
}
public static void main(String[] args) {
try {
List<Thread> jobs = Lists.newArrayList();
for (int i = 0; i < 10; i++) {
Thread t = new Thread(new ParallelJob("Parallel任务" + i, readLock));
jobs.add(t);
}
for (int i = 0; i < 10; i++) {
Thread t = new Thread(new MutexJob("Mutex任务" + i, writeLock));
jobs.add(t);
}
for (Thread t : jobs) {
t.start();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
CloseableUtils.closeQuietly(client);
}
}
}
看到没，用法和java concurrent包里的ReentrantReadWriteLock 是一模一样的。
事实上，整个recipes包的目录结构、实现原理同java concurrent包的设置是很一致的。比如有queue，Semaphore，Barrier等类，。他整个就是模仿jdk的实现，只不过是基于分布式的！
后边的几项，Barriers（关卡），Atomic（原子量），Caches，Queues和java concurrent包里的类的用法是一样的，就不继续贴了，有些附件里有。
要说明的是：有的功能性能不是特别理想，网上也没见有大的项目的使用案例。比如基于CAS机制的atomic，在某些情况重试的效率还不如硬同步，要是zookeeper节点再一多，各个节点之间通过event触发的数据同步极其频繁。那性能可以想象。
三、测试方法
curator提供了很好的测试工具，你甚至是可以在完全没有搭建zookeeper server端的情况下，完成测试。
有2个重要的类
TestingServer 模拟单点， TestingCluster模拟集群。
需要使用的话，得依赖
Xml代码
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<version>2.5.0</version>
</dependency>
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Curator是Netflix开源的一套ZooKeeper客户端框架. Netflix在使用ZooKeeper的过程中发现ZooKeeper自带的客户端太底层, 应用方在使用的时候需要自己处理很多事情
- Zookeeper开源客户端框架Curator简介与示例
- Zookeeper架构学习（三）:开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- zookeeper学习4之curator开源的zookeeper客户端
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- ZooKeeper(3.4.5) - 开源客户端 Curator(2.7.0) 的简单示例
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- Zookeeper开源客户端框架Curator简介
- 八：Zookeeper开源客户端Curator的api测试