Apache curator-recipes代码范例

我会带着你远行 2022-09-21 01:12 3阅读 0赞

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 Map<Thread,Boolean> lockedThread = new WeakHashMap<Thread,Boolean>();
    
        private String lockPath;
    
        private ConnectionStateListener stateListener = new StateListener();
    
        private RevocationListener<InterProcessMutex> revocationListener;
        public DistributedLock(CuratorFramework client,String path){
            lockPath = path;
            revocationListener = new RevocationListener<InterProcessMutex>() {
                @Override
                public void revocationRequested(InterProcessMutex forLock) {
                    if(!forLock.isAcquiredInThisProcess()){
                        return;
                    }
                    try{
                        forLock.release();
                    }catch(Exception e){
                        e.printStackTrace();
                    }
                }
            };
            lock = createLock(client);
            lock.makeRevocable(revocationListener);
            client.getConnectionStateListenable().addListener(stateListener);
        }
    
        public boolean lock(){
            try{
                lock.acquire();
                lockedThread.put(Thread.currentThread(),Boolean.TRUE);
            } catch (Exception e){
                //
            }
            return false;
        }
    
        public void unlock(){
            try{
                lock.release();
            }catch (Exception e){
                //
            }
        }
    
        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) {
                if(Boolean.FALSE.equals(lockedThread.get(Thread.currentThread()))){
                    return;//如果当前lock没有获取锁,则忽略
                }
                switch (newState){
                    case LOST:
                        //一旦丢失链接,就意味着zk server端已经删除了锁数据
                        lockedThread.clear();
                        lock = createLock(client);//must be rebuild
                        break;
                    default:
                        System.out.println(newState.toString());
    
                }
            }
        }
    }

? ? 底层的机制非常的简单: "获取锁"的操作,就是在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开发之旅更加愉快.

转：http://www.chengxuyuans.com/Java+/72042.html