Zookeeper--开源客户端Curator异步接口

爱被打了一巴掌 2021-08-28 12:31 393阅读 0赞

### Curator异步接口 ###

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

public interface BackgroundCallback { 
        void processResult(CuratorFramework client, CuratorEvent event) throws Exception;
    }

Backg roundCallback接口只有一一个processResult方法，该方法会在操作完成后被异步调用。该方法的参数说明如下：  
![在这里插入图片描述][20200512121132409.png]

CuratorEvent定义了Zookeeper服务端发送到客户端的一系列事件参数，其中比较重要的有事件类型和响应码两个参数：

事件类型（CuratorEventType）：

getType()会返回事件的类型，主要有：

*  CREATE：对应方法`CuratorFramework#create()`
 *  DELETE：对应方法`CuratorFramework#delete()`
 *  EXISTS：对应方法`CuratorFramework#checkExists()`
 *  GET\_DATA：`CuratorFramework#getData()`
 *  SET\_DATA：`CuratorFramework#setData()`
 *  CHILDREN：`CuratorFramework#getChildren()`
 *  SYNC：`CuratorFramework#sync(String,Object)`
 *  GET\_ACL：`CuratorFramework#getACL()`
 *  WATCHED：`Watchable#usingWatcher(Watcher)`和`Watchable#watched()`
 *  CLOSING：对应Zookeeper客户端与服务度断开连接事件

响应码（int）：  
响应码用于标识事件的结果状态，所有响应码都被定义在org . apache. zookeeper .KeeperException. Code类中

OK(0),//成功
            SYSTEMERROR(-1),//服务端内部错误
            RUNTIMEINCONSISTENCY(-2),
            DATAINCONSISTENCY(-3),
            CONNECTIONLOSS(-4),//断开连接
            MARSHALLINGERROR(-5),
            UNIMPLEMENTED(-6),
            OPERATIONTIMEOUT(-7),
            BADARGUMENTS(-8),
            APIERROR(-100),
            NONODE(-101),
            NOAUTH(-102),
            BADVERSION(-103),
            NOCHILDRENFOREPHEMERALS(-108),
            NODEEXISTS(-110),//指定节点已存在
            NOTEMPTY(-111),
            SESSIONEXPIRED(-112),//会话过期
            INVALIDCALLBACK(-113),
            INVALIDACL(-114),
            AUTHFAILED(-115),
            SESSIONMOVED(-118),
            NOTREADONLY(-119);

异步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);

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

//使用Curator的异步接口
    public class Create_Node_Background_Sample { 
    
        static String path = "/zk-book";
        static CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1: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 thread: " + Thread.currentThread().getName());
            //此处传入了自定的Executor
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
                    .inBackground(new BackgroundCallback() { 
                        @Override
                        public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception { 
                            System.out.println("event[code: ]" + curatorEvent.getResultCode() + ", type: " + curatorEvent.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 BackgroundCallback() { 
                        @Override
                        public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception { 
                            System.out.println("event[code: ]" + curatorEvent.getResultCode() + ", type: " + curatorEvent.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: ]-110, type: CREATE}
    Thread of processResult: main-EventThread
    event[code: ]0, type: CREATE}
    Thread of processResult: pool-3-thread-1

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

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

[20200512121132409.png]: /images/20210813/0195f9658bdd4e23bfbcc3128d8f39da.png