云原生 zookeeper应用之分布式队列

队列这种数据结构都不陌生，特点就是先进先出。有很多常用的消息中间件可以有现成的该部分功能，这里使用zookeeper基于发布订阅模式来实现分布式队列。对应的会有一个生产者和一个消费者。

这里理论上还是使用顺序节点。生产者不断产生新的顺序子节点，消费者watcher监听节点新增事件来消费消息。

生产者：

CuratorFramework client = ...

client.start();

String path = "/testqueue";

client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(path,"11".getBytes())；

消费者：

CuratorFramework client = ...

client.start();

String path = "/testqueue";

PathChildrenCache pathCache = new PathChildrenCache(client,path,true);

pathCache.getListenable().addListener(new PathChildrenCacheListener() {

@Override

public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {

if(event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED){

ChildData data = event.getData();

//handle msg

client.delete().forPath(data.getPath());

}

}

});

pathCache.start();

使用curator queue：

先来使用基本的队列类DistributedQueue。

DistributedQueue的初始化需要提交准备几个参数：

client连接就不多说了:

CuratorFramework client = ...

QueueSerializer：这个主要是用来指定对消息data进行序列化和反序列化

这里就搞一个简单的字符串类型：

QueueSerializer serializer = new QueueSerializer() {

@Override

public byte[] serialize(String item) {

return item.getBytes();

}

@Override

public String deserialize(byte[] bytes) {

return new String(bytes);

}

};

QueueConsumer消息consumer，当有新消息来的时候会调用consumer.consumeMessage()来处理消息

这里也搞个简单的string类型的处理consumer

QueueConsumer consumer = new QueueConsumer() {

@Override

public void consumeMessage(String s) throws Exception {

System.out.println("receive msg:"+s);

}

@Override

public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {

//TODO

}

};

队列消息发布：

//队列节点路径

String queuePath = "/queue";

//使用上面准备的几个参数构造DistributedQueue对象

DistributedQueue queue = QueueBuilder.builder(client,consumer,serializer,queuePath).buildQueue();

queue.start();

//调用put方法生产消息

queue.put("hello");

queue.put("msg");

Thread.sleep(2000);

queue.put("3");

这样在启动测试程序在，consumer的consumeMessage方法就会收到queue.put的消息。

这里有个问题有没有发现，在初始化queue的时候需要指定consumer，那岂不是只能同一个程序中生产消费，何来的分布式？

其实这里在queue对象创建的时候consumer可以为null，这个时候queue就只生产消息。具体的逻辑需要看下DistributedQueue类的源码。

在DistributedQueue类的构造函数有一步设置isProducerOnly属性

isProducerOnly = (consumer == null);

然后在start()方法会根据isProducerOnly来判断启动方式

if ( !isProducerOnly || (maxItems != QueueBuilder.NOT_SET) )

{

childrenCache.start();

}

if ( !isProducerOnly )

{

service.submit

(

new Callable