Zookeeper应用之——队列(Queue)

为了在Zookeeper中实现分布式队列,首先需要设计一个znode来存放数据,这个节点叫做队列节点,我们的例子中这个节点是/zookeeper/queue。 生产者向队列中存放数据,每一个消息都是队列节点下的一个新节点,叫做消息节点。消息节点的命名规则为:queue-xxx,xxx是一个单调 递增的序列,我们可以在创建节点时指定创建模式为PERSISTENT_SEQUENTIAL来实现。这样,生产者不断的向队列节点中发送消息,消息为queue-xxx, 队列中,生产者这一端就解决了,我们具体看一下代码:

Producer(生产者)

Zookeeper应用之——队列(Queue)

生产者每隔10s向队列中存放消息,消息节点的类型为PERSISTENT_SEQUENTIAL,消息节点中的数据为Thread.currentThread().getName()+”-“+i。

消费者

消费者从队列节点中获取消息,我们使用getChildren()方法获取到队列节点中的所有消息,然后获取消息节点数据,消费消息,并删除消息节点。 如果getChildren()没有获取到数据,说明队列是空的,则消费者等待,然后再调用getChildren()方法设置观察者监听队列节点,队列节点发生变化后 (子节点改变),触发监听事件,唤起消费者。消费者实现如下:

Zookeeper应用之——队列(Queue)

Zookeeper应用之——队列(Queue)

Zookeeper应用之——队列(Queue)

上面的例子中有一个局限性,就是 消费者只能有一个 。队列的用户有两个:广播和队列。

  • 广播是所有消费者都拿到消息并消费,我们的例子在删除消息节点时,不能保证其他消费者都拿到了这个消息。
  • 队列是一个消息只能被一个消费者消费,我们的例子中,消费者获取消息时,并没有加锁。

所以我们只启动一个消费者来演示,主函数如下:

Zookeeper应用之——队列(Queue)

后台打印结果如下:

Zookeeper应用之——队列(Queue)

Zookeeper实现队列就介绍完了。

Zookeeper应用之——队列(Queue)


分享到:


相關文章: