简介

curator实现了先入先出的分布式消息队列，它采用的是zookeeper的持久化有序节点。

官方文档：

javaDoc：

代码示例

import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.recipes.queue.DistributedQueue;import org.apache.curator.framework.recipes.queue.QueueBuilder;import org.apache.curator.framework.recipes.queue.QueueConsumer;import org.apache.curator.framework.recipes.queue.QueueSerializer;import org.apache.curator.framework.state.ConnectionState;import org.apache.curator.retry.ExponentialBackoffRetry;import java.io.UnsupportedEncodingException;public class DistributedQueueDemo {    private static CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(3000, 2));    private static String path = "/queue/0001";    private static DistributedQueue
     
       queue;    static {        client.start();    }    public static void main(String[] args) throws Exception {        QueueConsumer
      
        consumer = new QueueConsumer
       
        () {            @Override            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {                System.out.println("state changed");            }            @Override            public void consumeMessage(String s) {                System.out.println("消费数据：" + s);            }        };        queue = QueueBuilder.builder(client, consumer, new QueueSerializer
        
         () {            @Override            public byte[] serialize(String s) {                try {                    return s.getBytes("utf-8");                } catch (UnsupportedEncodingException e) {                    e.printStackTrace();                }                return null;            }            @Override            public String deserialize(byte[] bytes) {                return new String(bytes);            }        }, path).buildQueue();        queue.start();        System.out.println("queue built");        queue.put("a");        queue.put("b");        System.out.println("put ended");        Thread.sleep(100);        queue.close();        Thread.sleep(50000);        client.close();    }}