Apache curator-recipes代码实例

    Apache curator-recipes组件提供了大量已经"生产化"(produced)的特性,极大的简化了使用zk的复杂度.

    1. Cache: 提供了对一个Node持续监听,如果节点数据变更,即可立即得到响应. 开发者无需过度的关注watcher和Event操作.

    2. Queues: 提供了重量级的分布式队列解决方案,比如:权重队列,可延迟队列等.其实Zookeeper并不适合作为数据存储系统,你可以适度的使用它来达成分布式队列的设计要求.

    3. Counters: 全局计数器是分布式设计中很常用的,包括"全局计算器"和"原子自增计数器".

    4. Locks: 分布式锁的设计有很多手段,此组件提供了分布式"读写分写锁"/"共享锁"等.在我们需要控制资源访问的情况下,非常有用.

    5. Barries: 栅栏,需要对分布式环境中,多个操作(进程)进行同步或者协同时,可以考虑使用barries;

    6. Elections: 选举;可以在多个"注册者"之间选举出leader,作为操作调度/任务监控/队列消费的执行者,我们在设计"leader角色选举"/"单点任务执行""分布式队列消费者"等场景时,非常有效.

 

    代码实例

1. 创建Client

CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
//fluent style
String namespace = "cluster-worker";
CuratorFramework client = builder.connectString("127.0.0.1:2181")
		.sessionTimeoutMs(30000)
		.connectionTimeoutMs(30000)
		.canBeReadOnly(false)
		.retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE))
		.namespace(namespace)
		.defaultData(null)
		.build();
client.start();
EnsurePath ensure = client.newNamespaceAwareEnsurePath(namespace);
//code for test
Thread.sleep(5000);
//client.close()//

 

2. Caches

    持续watcher节点,并将节点的数据变更即时的在本地反应出来.recpise提供了PathChildrenCache和NodeCache两个API.

public static PathChildrenCache pathChildrenCache(CuratorFramework client, String path, Boolean cacheData) throws Exception {
	final PathChildrenCache cached = new PathChildrenCache(client, path, cacheData);
	cached.getListenable().addListener(new PathChildrenCacheListener() {
		@Override
		public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
			PathChildrenCacheEvent.Type eventType = event.getType();
			switch (eventType) {
				case CONNECTION_RECONNECTED:
					cached.rebuild();
					break;
				case CONNECTION_SUSPENDED:
				case CONNECTION_LOST:
					System.out.println("Connection error,waiting...");
					break;
				default:
					System.out.println("Data:" + event.getData().toString());
			}
		}
	});
	return cached;
}

 

PathChildrenCache cached = pathChildrenCache(client,path,true);
//= start() + rebuild()
//事件操作,将会在额外的线程中执行.
cached.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
List<ChildData> childData = cached.getCurrentData();
if (childData != null) {
	for (ChildData data : childData) {
		System.out.println("Path:" + data.getPath() + ",data" + new String(data.getData(), "utf-8"));
	}
}

//当不在需要关注此节点数据时,需要及时的关闭它.
//因为每个cached,都会额外的消耗一个线程.
//cached.close();////close the watcher,clear the cached Data

    对于PathChildrenCache.getCurrentData()将从获取本地的数据列表,而不是触发一次zookeeper.getChildren(),因此为"Cache".

3. Queues:分布式队列

    分布式队列的基本特性,就是"生产者"或"消费者"跨越多个进程,且在此种环境中需要确保队列的push/poll的有序性.

    zookeeper本身并没有提供分布式队列的实现,只是recipse根据zookeeper的watcher和具有version标记的node,来间接的实现分布式queue..内部机制如下:

    --> 如果是消费者(QueueConsumer),会创建一个类似于PathChildrenCache的实例用于监听queuePath下的子节点变更事件(单独的线程中).同时consumer处于阻塞状态,当有子节点变更事件时会被唤醒(包括创建子节点/删除子节点等);

    --> 此时consumer获取子节点列表,并将每个节点信息封装成Runnable任务单元,提交到线程池中. 

    --> Runnable中并执行QueueConsumer.consumer(.)方法.

    --> 如果是生产者,则发布一个message时recipes将会在queuePath下创建一个PERSISTENT_SEQUENTIAL节点,同时保存message数据. 消费时,也将按照节点的顺序进行.

 

    发布消息并没有太多的问题仅仅是创建一个"有序"节点即可..但是对于消费者,那么需要考虑的因数就很多,比如:1) 多个消费者同时消费时,需要确保消息不能重复且有序 2) 消息消费时,如果网络异常,怎么办?

    对于QistributedQueue中,对上述问题的解决办法也非常粗糙,内部机制如下:

    --> 如果使用了消费担保(即指定了lockPath),在调用consumer方法之前,首先创建一个临时节点(lockPath + 子节点),如果创建此临时节点失败也就意味着此消息被其他消费者,则忽略此消息.

    --> 然后从子节点中获取数据,如果获取失败,意味着此节点已经被其他消费者删除,则忽略此消息.

    --> 然后调用consumer()方法,如果此方法抛出异常,消息将会再次添加到队列中(删除旧的子节点,创建一个新的子节点).如果消费正常,则删除节点.

    --> 无论成败,则删除临时节点(lockPath + 子节点)

 

    --> 如果没有使用消费担保,则首先获取子节点的数据(getData),然后立即删除此子节点

    -->调用consumer()方法.

 

    问题:

    1) 源代码中,在获取子节点的data时(getData)并没有指定version校验,我深度怀疑,当消息被并发的消费时,是否有重复的可能.

    2) 因为zookeeper本身获取一个节点的子节点列表时,将得到所有的子节点,那么就意味任何一个消费者中每次Event触发,都将获取整个childrenList,如果此列表很庞大,性能问题将是非常突出的.

    3) 在消费担保的情况下,每消费一个消息,就会做"创建临时节点""删除临时节点""获取数据"等大量工作,如果有多个消费者同时运行,那么对zk的操作次数将会倍数级增加,性能问题以及数据安全性问题,也是非常值得考虑的.

    4) zookeeper已经限定了每个节点的数据尺寸,以及每个节点下子节点的个数,这对于实现规模性的分布式队列,确实不是良好的选择.

    5) 当队列中,消费者和生产者的速率不均衡时,问题将会更加严重,比如:快速的生产者 + 慢速的消费者;因为此时Event事件将会非常频繁(网络消耗严重),对于消费者而言,线程上线切换锁带来的性能消耗,不可忽视.

 

    最终,我们需要明确使用zookeeper作为分布式队列的场景: 1) 队列深度较小 2) 生产者和消费者的速度都非常的低且消费者消费速度更快,即单位时间内产生的消息很少.  3) 建议只有一个消费者.

    比如: 一个数据分析系统,这个系统中有多个定时任务,当任务即将触发时,向此分布式队列中提交一个消息,消息中包含任务的ID,那么谁消费了此任务ID,那么谁就负责执行此任务.我们间接的实现了分布式任务单点运行的需求.

    还有一种场景,比如实现"排它重入锁"也可以使用DistributedQueue作为底层支撑.

 

    消费者:DistributedQueueConusumer.java

public class DistributedQueueConsumer {

    private DistributedQueue<String> queue;

    public DistributedQueueConsumer(CuratorFramework client, String queuePath) throws Exception {
        QueueBuilder<String> builder = QueueBuilder.builder(client, new StringQueueConsumer(), new StringQueueSerializer(), queuePath);
        queue = builder.lockPath("queue-lock").buildQueue();//消费担保
    }

    public void start() throws Exception {
        queue.start();
    }

    public void close() throws Exception {
        queue.close();
    }

    public static DistributedQueue distributedQueueAsProducer(CuratorFramework client, String path) throws Exception {
        QueueBuilder<String> builder = QueueBuilder.builder(client, null, new StringQueueSerializer(), path);
        builder.maxItems(1024);// 有界队列,最大队列深度,如果深度达到此值,将阻塞"生产者"创建新的节点.
        return builder.buildQueue();
    }

    //utils for producer and consumer
    static class StringQueueSerializer implements QueueSerializer<String> {
        private static final Charset charset = Charset.forName("utf-8");

        //as producer
        @Override
        public byte[] serialize(String item) {
            return item.getBytes(charset);
        }

        //as consumer
        @Override
        public String deserialize(byte[] bytes) {
            return new String(bytes, charset);
        }
    }


    class StringQueueConsumer implements QueueConsumer<String> {
        static final String TAG = "_TAG";
        @Override
        public void consumeMessage(String message) throws Exception {
            System.out.println("Consumer:" + message);
            if (message.equals(TAG)) {
                System.out.println("Tag message...ignore it.");
            }
        }

        @Override
        public void stateChanged(CuratorFramework client, ConnectionState newState) {
            switch (newState) {
                case RECONNECTED:
                    try {
                        //当链接重建之后,需要发送一个TAG消息,用于重新触发本地的watcher,以便获取新的children列表
                        queue.put(TAG);
                    } catch (Exception e) {
                        //
                    }
                    break;
                default:
                    System.out.println(newState.toString());
            }
        }
    }

}

    生产者: DistributedQueueProducer.java

public class DistributedQueueProducer {

    private DistributedQueue<String> queue;

    public DistributedQueueProducer(CuratorFramework client, String queuePath) throws Exception {
        QueueBuilder<String> builder = QueueBuilder.builder(client, null, new StringQueueSerializer(), queuePath);
        queue = builder.lockPath("queue-lock").buildQueue();//消费担保
    }

    public void start() throws Exception {
        queue.start();
    }

    public void close() throws Exception {
        queue.close();
    }

    public void put(String message) throws  Exception{
        queue.put(message);
    }

}

     

    Recipse还提供了其他2个API:

    1) DistributedIdQueue: 内部基于DistributedQueue的所有机制,只是除了指定queue中消息的内容之外,还可以指定一个ID,这个ID作为消息的标记,最终此ID值将作为znode的path后缀.此后可以通过ID去消费(dequeue)一个消息.队列的排序方式是根据ID的字典顺序--正序.

    2) DistributedProrityQueue: 有权重的队列,内部基于DistributedQueue,不过在发布消息时,需要指定此消息的权重数字;队列的排序方式为根据权重排序.

  

    此外DistributedQueue的开发中,必须在QueueConsumer中关注"链接失效"的事件.

 

4. Counters

    计数器,其中SharedCount可以用来监听zookeeper中一个Integer类型的数字变更.此外还有一个更加基础的API: SharedValue,你可以实现任意类型的计数器.

final SharedCount counter = new SharedCount(client,"/counter-vv",0);
counter.addListener(new SharedCountListener(){
	@Override
	public void countHasChanged(SharedCountReader sharedCount, int newCount) throws Exception {
		//
		System.out.println("count changed:" + newCount);
	}

	@Override
	public void stateChanged(CuratorFramework client, ConnectionState newState) {
		switch (newState){
			case RECONNECTED:
				try {
					//当链接重建之后,需要手动fresh
					Integer current = counter.getCount();
					counter.trySetCount(current);//reflush,无论更新成败,都会获取最新的值
				} catch (Exception e) {
					//
				}
				break;
			default:
				System.out.println(newState.toString());
		}
	}
});
counter.start();

//counter.close();//取消watcher

    SharedCount中也需要关注"链接异常"的问题,我们可以通过注册listener的方式,当链接重连成功后,重新获取新的值.

    在"计数器"中,还提供了DistributedAtomicInteger,DistributedAtomicLong两个分布式自增计数器.

DistributedAtomicInteger atomicInteger = new DistributedAtomicInteger(client,"/counter-vv",new RetryNTimes(32,1000));
AtomicValue<Integer> rc = atomicInteger.increment();
System.out.println("success:" + rc.succeeded() + ";before:" + rc.preValue() + ";after:" + rc.postValue());
//atomicInteger.add(1); //++
//atomicInteger.subtract(1); //--

 

5. Locks 

    使用zookeeper作为分布式锁,是一个普遍的需求,如下展示如何设计一个分布式重入锁,其中一个path表示一个锁资源.

public class DistributedLock{

    private InterProcessMutex lock;//重入的,排他的.
    
    private LockHolder _holder;

    private String lockPath;

    private ConnectionStateListener stateListener = new StateListener();
    
    private LockEvent lockEvent;
    
    private Semaphore semaphore = new Semaphore(0);
    private RevocationListener<InterProcessMutex> revocationListener;
    
    public DistributedLock(CuratorFramework client,String path,final LockEvent lockEvent){
        lockPath = path;
        this.lockEvent = lockEvent;
        revocationListener = new RevocationListener<InterProcessMutex>() {
            @Override
            public void revocationRequested(InterProcessMutex forLock) {
                if(!forLock.isAcquiredInThisProcess()){
                    return;
                }
                try{
                	lockEvent.beforeRelease();
                	//只有当前线程才可以释放
                	//其他线程,将会抛出一个错误.
                    forLock.release();
                }catch(Exception e){
                    e.printStackTrace();
                }
            }
        };
        lock = createLock(client);
        lock.makeRevocable(revocationListener);
        client.getConnectionStateListenable().addListener(stateListener);
    }
    

    
    public boolean lock(){
        try{
        	if(_holder == null) {
        		_holder = new LockHolder();
            	_holder.setDaemon(true);
        	}
            if(!_holder.isAlive()){
            	_holder.start();
            }
            semaphore.acquire();//直到_holder正常运行
        } catch (Exception e){
            //
        }
        return false;
    }

    public void unlock(){
        if(_holder.isAlive()){
        	_holder.interrupt();
        }
    }
    

    private InterProcessMutex createLock(CuratorFramework client){
        lock = new InterProcessMutex(client,lockPath);
        //协同中断,如果其他线程/进程需要此锁中断时,调用此listener.
        lock.makeRevocable(revocationListener);
        client.getConnectionStateListenable().addListener(stateListener);
        return lock;
    }

    class StateListener implements ConnectionStateListener{
        @Override
        public void stateChanged(CuratorFramework client, ConnectionState newState) {
            switch (newState){
                case LOST:
                    //一旦丢失链接,就意味着zk server端已经删除了锁数据
                    boolean rebuild = lockEvent.lose();
                    _holder.interrupt();//NUll
                    if(rebuild){
                    	lock();
                    }
                    break;
                default:
                    System.out.println(newState.toString());

            }
        }
    }
    
    static interface LockEvent {
    	
    	public void afterAquire();
    	
    	public boolean lose();
    	
    	//释放锁
    	public void beforeRelease();
    }
    
    class LockHolder extends Thread{
		@Override
		public void run() {
			try{
				lock.acquire();
	            semaphore.release();//
	            lockEvent.afterAquire();
	            synchronized (lock) {
					lock.wait();
				}
			}catch(InterruptedException e){
				try{
					lockEvent.beforeRelease();
					lock.release();
				}catch(Exception ie){
					ie.printStackTrace();
				}
			}catch (Exception e) {
				e.printStackTrace();
			} 
			
		}
		
    	
    }
    
    public static void main(String[] args) throws Exception {
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
        //fluent style
        String namespace = "cluster-worker";
        CuratorFramework client = builder.connectString("127.0.0.1:2181")   //FixedEnsembleProvider
                .sessionTimeoutMs(30000)
                .connectionTimeoutMs(30000)
                .canBeReadOnly(false)   // cant connect to one observer-instance
                .retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE)) //auto reconnect policy
                .namespace(namespace)  //good method,you can specify one fix prefiex-path of all znode.
                .defaultData(null)
                .build();
        client.start();
        EnsurePath ensure = client.newNamespaceAwareEnsurePath(namespace);
        LockEvent le = new LockEvent() {
			
			@Override
			public boolean lose() {
				System.out.println("lose,shoud be waiting or stop workers!");
				return false;
			}
			
			@Override
			public void beforeRelease() {
				System.out.println("Lock would be released!");
				
			}
			
			@Override
			public void afterAquire() {
				System.out.println("Locked success,can be running workers");
				
			}
		};
        DistributedLock distLock = new DistributedLock(client,"/lock",le);
        distLock.lock();
        distLock.unlock();
        Thread.sleep(2000);
        client.close();

    }

}

    底层的机制非常的简单: "获取锁"的操作,就是在zookeeper中创建一个EPHEMERAL_SEQUENTIAL节点,同时对此节点的临近节点注册一个watcher;当"临近节点"被删除时,表示其他进程已经释放了锁,此watcher将会触发,并唤醒当前线程,然后acquire方法返回.."释放锁"的操作,就是删除此临时节点.此时临近的下一个节点将获得锁..

    所谓"重入",就是同一个线程多次获取锁时,如果此线程已经持有了锁(即创建了zk临时节点),事实上将不会再次创建zk的临时节点,而是直接返回.

    因为"重入锁",基于临时节点的特性,因此必须关注client链接重建的问题;粗糙的解决办法,就是每次链接重建(session过期),重新实例化lock对象. 

 

6. Barrier

    栅栏, 可以用来协同分布式环境中的线程.让他们有条件的阻塞,且同时唤醒.

DistributedBarrier barrier = new DistributedBarrier(client,"/barrier");
barrier.setBarrier(); //设置barrier
System.out.println("setBarrier...");
barrier.waitOnBarrier();//等待其他进程移除barrier,此后所有的waitOnBarrier进程都将解除阻塞.
//barrier.removeBarrier(); //移除barrier,解除阻塞.

DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(client,"/d-barrier",12);
System.out.println("enter...");
barrier.enter();//阻塞,直到12个成员加入
System.out.println("running...");
barrier.leave();//阻塞,直到12个成员离开

    其中DistributedDoubleBarrier为双端栅栏,可以让N个线程(进程)同时开始,并且同时退出..

    对于DistributedBarrier内部机制非常简单: setBarrier()方法就是创建"栅栏"节点,removeBarrier()方法就是删除此节点;当执行setBarrier之后,所有的waitOnBarrier()操作都将阻塞,直到删除节点的事件触发.

 

7.LeaderSelector

    在很多场景中,我们需要"leader选举";比如在分布式有很多task,这些task的执行时机需要一个"Leader"去调度.任何时候,同一个leaderPath节点下,只会有一个"leader"..如下展示如何简单的使用LeaderSelector.

public class LeaderSelectorClient {

    private LeaderSelector selector;

    private final Object lock = new Object();
    private boolean isLeader = false;

    public LeaderSelectorClient(CuratorFramework client, String leaderPath) throws Exception {
        LeaderSelectorListener selectorListener = new LeaderSelectorListener() {
            //此方法将会在Selector的线程池中的线程调用
            @Override
            public void takeLeadership(CuratorFramework client) throws Exception {
                System.out.println("I am leader...");
                //如果takeLeadership方法被调用,说明此selector实例已经为leader
                //此方法需要阻塞,直到selector放弃leader角色
                isLeader = true;
                while (isLeader) {
                    synchronized (lock) {
                        lock.wait();
                    }
                }
            }

            //这个方法将会在Zookeeper主线程中调用---watcher响应时
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.out.println("Connection state changed...");
                //对于LeaderSelector,底层实现为对leaderPath节点使用了"排他锁",
                //"排他锁"的本质,就是一个"临时节点"
                //如果接收到LOST,说明此selector实例已经丢失了leader信息.
                if (newState == ConnectionState.LOST) {
                    isLeader = false;
                    synchronized (lock) {
                        lock.notifyAll();
                    }
                }
            }
        };
        selector = new LeaderSelector(client, leaderPath, selectorListener);
        //一旦leader释放角色之后,是否继续参与leader的选举
        //此处需要关注CuratorFrameworker.RetryPolicy策略.
        //1) 如果leader是耐久性的,selector实例需要一致关注leader的状态,可以autoRequeue
        //2) 如果leader再行使完任务之后,释放,然后在此后的某个时刻再次选举(比如定时任务),此处可以保持默认值false
        selector.autoRequeue();
    }


    public void start() {
        selector.start();
    }

    public void release() {
        //释放leader角色
        isLeader = false;
        //takeLeadership方法将会中断并返回.
        selector.interruptLeadership();
        synchronized (lock){
            lock.notifyAll();//
        }
    }

    //重新获取leader角色--选举
    public void take() {
        selector.requeue();
    }

    public void close() {
        isLeader = false;
        selector.close();
        synchronized (lock){
            lock.notifyAll();//
        }
    }

    public boolean isLeader() {
        return selector.hasLeadership();
    }

}

    开发者使用LeaderSelector时,需要关注takeLeadership方法的内部逻辑,一旦takeLeadership方法被调用,那么此selector已经是leader角色了,你可以在此方法中增加"事件通知"等来执行一些异步的操作.

    isLeader()方法只能返回当前的状态,有可能返回true之后不久,这个selector实例将不不再是leader,那么就需要我们在listener中更多的关注stateChanged过程.

 

8. LeaderLatch

    上述LeaderSelector开发中,开发者需要关注"链接异常"情况,也需要自己去阻塞leader角色变更,也需要自己去封装"leader角色变更"时的事件处理器....recipse组件已经实现了LeaderLatch来初步解决上述问题,这个类是个便捷的类,如果你还需要更多的处理,恐怕还是需要自己去封装LeaderSelector.

LeaderLatch latch = new LeaderLatch(client,"/task/leader");
//让listener在单独的线程池中运行
Executor executor = Executors.newCachedThreadPool();
//每个listener都用来执行角色变换的事件处理.
LeaderLatchListener latchListener = new LeaderLatchListener() {
	@Override
	public void isLeader() {
		System.out.println("I am leader...");
	}

	@Override
	public void notLeader() {
		System.out.println("I am not leader...");
	}
};
latch.addListener(latchListener,executor);
latch.start();
latch.await();//等待leader角色.
//在await退出之后,你需要通过其他手段继续关注leader状态变更.
System.out.println(latch.hasLeadership());
Thread.sleep(5000);
latch.close();
Thread.sleep(2000);
client.close();

 

  

    到此为止,我们已经把curator-recipse的大部分API都介绍完毕了,希望我们的zookeeper开发之旅更加愉快.

    

评论

4 楼 QING____ 2014-04-20  

liupengjava 写道

//协同中断,如果其他线程/进程需要此锁中断时,调用此listener. 
需要用静态方法：Revoker.attemptRevoke(client, lock.getLockPath());
getLockPath()是一个锁的路径，必须要提前写到一个zk文件中，要不然跨jvm是没办法获取的吧~。。

非常感谢你的提问,这个段程序确实有些问题,不过后来我们已经调整了对分布式锁的使用方式,部分问题已经解决,我已经调整了博文中关于DistributedLock的源码(实际环境代码的缩减版),请你参考.

Lock不能跨线程release.

3 楼 liupengjava 2014-04-18  

//协同中断,如果其他线程/进程需要此锁中断时,调用此listener. 
需要用静态方法：Revoker.attemptRevoke(client, lock.getLockPath());
getLockPath()是一个锁的路径，必须要提前写到一个zk文件中，要不然跨jvm是没办法获取的吧~。。

2 楼 liupengjava 2014-04-18  

//协同中断,如果其他线程/进程需要此锁中断时,调用此listener. 

1 楼 liupengjava 2014-04-18  

RevocationListener里面获取到forLock之后，调用forLock.release();  会报错，因为线程不一样了，就不能release了。