Curator之异步接口。

亦凉 2024-02-18 17:07 16阅读 0赞

Curator中引入了BackgroundCallback接口，用来处理异步接口调用之后服务端返回的结果信息，其接口定义如下。

public interface BackgroundCallback \{

/\*\*

\*Called when the async background operation completes

\*@param client the client

\*@param event operation result details

\*@throws Exception errors

\*/

public void processResult(CuratorFramework client, CuratorEvent event) throws Exception;

BackgroundCallback接口只有一个processResult方法，从注释中可以看出，该方法会在操作完成后被异步调用。该方法的参数说明如下表所示。

<table> 
 <tbody> 
  <tr> 
   <th>参数名</th> 
   <th>说明</th> 
  </tr> 
 </tbody> 
 <tbody> 
  <tr> 
   <td>client</td> 
   <td>当前客户端实例</td> 
  </tr> 
  <tr> 
   <td>event</td> 
   <td>服务端事件</td> 
  </tr> 
 </tbody> 
</table>

对于BackgroundCallback接口，我们重点来看CuratorEvent这个参数。CuratorEvent定义了ZooKeeper服务端发送到客户端的一系列事件参数，其中比较重要的有事件类型和响应码两个参数。

## 事件类型（CuratorEventType） ##

getType()，代表本次事件的类型，主要有CREATE、DELETE、EXISTS、GET\_DATA、SET\_DATA、CHILDREN、SYNC、GET\_ACL、WATCHED和CLOSING，分别代表CuratorFramework\#create()、CuratorFramework\#delete()、Curator Framework\#checkExists()、CUratorFramework\#getData()、CuratorFramework\#setData()、CuratorFramework\#getChildren()、CuratorFramework\#sync(String, Object)、CuratorFramework\#getACL()、Watchable\#using Watcher(Watcher)/ Watchable\# watched()和ZooKeeper客户端与服务端连接断开事件。

## 响应码（int） ##

响应码用于标识事件的结果状态，所有响应码都被定义在org.apache.zookeeper.KeeperException.Code类中，比较常见的响应码有0（Ok）、-4（ConnectionLoss）、-110NodeExists（）和-112（SessionExpired）等，分别代表接口调用成功、客户端与服务端连接已断开、指定节点已存在和会话已过期等。

在程序中，我们可以通过以下API来进行异步操作。

**Backgroundable<T>**

*  **public T inBackground();**
 *  **public T inBackground(Object context);**
 *  **public T inBackground(BackgroundCallback callback);**
 *  **public T inBackground(BackgroundCallback callback, Object context);**
 *  **public T inBackground(BackgroundCallback callback, Executor executor);**
 *  **public T inBackground(BackgroundCallback callback, Object context, Executor executor);**

在这些API接口中，我们重点来关注下executor这个参数。在ZooKeeper中，所有异步通知事件处理都是由EventThread这个线程来处理的——EventThread线程用于串行处理所有的事件通知。EventThread的“串行处理机制”在绝大部分应用场景下能够保证对事件处理的顺序性，带这个特性也有其弊端，就是一旦碰上一个复杂的处理单元，就会消耗过长的处理时间，从而影响对其他事件的处理。因此，在上面的inBackground接口中，允许用户传入一个Executor实例，这样一来，就可以把那些比较复杂的的事件处理放到一个专门的线程池中去，如Executors.newFixedThreadPool(2)。

下面，我们通过一个实际例子来看看如何使用Curator的异步接口。

// 使用Curator的异步接口

public class Create\_Node\_Background\_Sample \{

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) throws Exception \{

client.start();

System.out.println("Main thred:" + Thread.currentThread().getName());

// 此处传入了自定义的Executor

client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallbak()\{

@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();

\}, tp).forPath(path, "init".getBytes());

// 此处没有传入自定义的Executor

client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallbak()\{

@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();

运行程序，输出结果如下：

![20171212201644164][]

上面这个程序使用了异步接口inBackground来创建节点，前后两次调用，创建的节点名相同。从两次返回的event中可以看出，第一次返回的响应码是0，表明此次调用成功，即创建节点成功；而第二次返回的响应码是-110，表明该节点已经存在，无法重复创建。这些响应码和ZooKeeper原生的响应码是一致的。

另外，我们再来看看前后两次调用inBackground接口时传入的Executor参数。第一次传入了一个ExecutorService，这样一来，Curator的异步事件处理逻辑就会交由该线程池去做。而第二次调用时，没有传入任何Executor，因此会使用ZooKeeper默认的EventThread来处理。

[20171212201644164]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/7ab0360fcbde4010a3d31d46d95b90a3.png