【欢迎关注公众号：程序猿讲故事 （codestory），及时接收最新文章】

生产-消费者队列，用于多节点的分布式数据结构，生产和消费数据。生产者创建一个数据对象，并放到队列中；消费者从队列中取出一个数据对象并进行处理。在ZooKeeper中，队列可以使用一个容器节点下创建多个子节点来实现；创建子节点时，CreateMode使用 PERSISTENT_SEQUENTIAL，ZooKeeper会自动在节点名称后面添加唯一序列号。EPHEMERAL_SEQUENTIAL也有同样的特点，区别在于会话结束后是否会自动删除。

敲小黑板：*_SEQUENTIAL是ZooKeeper的一个很重要的特性，分布式锁、选举制度都依靠这个特性实现的。

1      对前续代码的重构

之前的文章，我们已经用实现了Watcher和Barrier，创建ZooKeeper连接的代码已经复制了一遍。后续还需要类似的工作，因此先对原有代码做一下重构，让代码味道干净一点。

以下是 process(WatchedEvent)的代码

final public void process(WatchedEvent event) {

if (Event.EventType.None.equals(event.getType())) {

// 连接状态发生变化

if (Event.KeeperState.SyncConnected.equals(event.getState())) {

// 连接建立成功

connectedSemaphore.countDown();

}

} else if (Event.EventType.NodeCreated.equals(event.getType())) {

processNodeCreated(event);

} else if (Event.EventType.NodeDeleted.equals(event.getType())) {

processNodeDeleted(event);

} else if (Event.EventType.NodeDataChanged.equals(event.getType())) {

processNodeDataChanged(event);

} else if (Event.EventType.NodeChildrenChanged.equals(event.getType())) {

processNodeChildrenChanged(event);

}

}

以ZooKeeperBarrier为例，看看重构之后的构造函数和监听Event的代码

ZooKeeperBarrier(String address, String tableSerial, int tableCapacity, String customerName)

throws IOException {

super(address);

this.tableSerial = createRootNode(tableSerial);

this.tableCapacity = tableCapacity;

this.customerName = customerName;

}

protected void processNodeChildrenChanged(WatchedEvent event) {

log.info("{} 接收到了通知 : {}", customerName, event.getType());

// 子节点有变化

synchronized (mutex) {

mutex.notify();

}

}

2      队列的生产者

生产者的关键代码

String elementName = queueName + "/element";

ArrayList<ACL> ids = ZooDefs.Ids.OPEN_ACL_UNSAFE;

CreateMode createMode = CreateMode.PERSISTENT_SEQUENTIAL;

getZooKeeper().create(elementName, value, ids, createMode);

注意，重点是PERSISTENT_SEQUENTIAL，PERSISTENT是表示永久存储直到有命令删除，SEQUENTIAL表示自动在后面添加自增的唯一序列号。这样，尽管elementName都一样，但实际生成的zNode名字在 “element”后面会添加格式为%010d的10个数字，如0000000001。如一个完整的zNode名可能为/queue/element0000000021。

3      测试日志

把测试结果放源码前面，免得大家被无聊的代码晃晕。

测试代码创建了两个线程，一个线程是生产者，按随机间隔往队列中添加对象；一个线程是消费者，随机间隔尝试从队列中取出第一个，如果当时队列为空，会等到直到新的数据。

两个进程都加上随机间隔，是为了模拟生产可能比消费更快的情况。以下是测试日志，为了更突出，生产和消费的日志我增加了不同的文字样式。

49:47.866 [INFO] ZooKeeperQueueTest.testQueue(29) 开始ZooKeeper队列测试，本次将测试 10 个数据

49:48.076 [DEBUG] ZooKeeperQueue.log(201)

+ Profiler [tech.codestory.zookeeper.queue.ZooKeeperQueue 连接到ZooKeeper]

|-- elapsed time                   [开始链接]   119.863 milliseconds.

|-- elapsed time           [等待连接成功的Event]    40.039 milliseconds.

|-- Total        [tech.codestory.zookeeper.queue.ZooKeeperQueue 连接到ZooKeeper]   159.911 milliseconds.

49:48.082 [DEBUG] ZooKeeperQueue.log(201)

+ Profiler [tech.codestory.zookeeper.queue.ZooKeeperQueue 连接到ZooKeeper]

|-- elapsed time                   [开始链接]   103.795 milliseconds.

|-- elapsed time           [等待连接成功的Event]    65.899 milliseconds.

|-- Total        [tech.codestory.zookeeper.queue.ZooKeeperQueue 连接到ZooKeeper]   170.263 milliseconds.

49:48.102 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 1 , 然后等待 1700 毫秒

49:48.134 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 1 , 然后等待 4000 毫秒

49:49.814 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 2 , 然后等待 900 毫秒

49:50.717 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 3 , 然后等待 1300 毫秒

49:52.020 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 4 , 然后等待 3700 毫秒

49:52.139 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 2 , 然后等待 2800 毫秒

49:54.947 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 3 , 然后等待 4500 毫秒

49:55.724 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 5 , 然后等待 3500 毫秒

49:59.228 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 6 , 然后等待 4200 毫秒

49:59.454 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 4 , 然后等待 2400 毫秒

50:01.870 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 5 , 然后等待 4900 毫秒

50:03.435 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 7 , 然后等待 4500 毫秒

50:06.776 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 6 , 然后等待 3600 毫秒

50:07.938 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 8 , 然后等待 1900 毫秒

50:09.846 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 9 , 然后等待 3200 毫秒

50:10.388 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 7 , 然后等待 2900 毫秒

50:13.051 [INFO] ZooKeeperQueueTest.run(51) 生产对象 : 10 , 然后等待 4900 毫秒

50:13.294 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 8 , 然后等待 300 毫秒

50:13.600 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 9 , 然后等待 4800 毫秒

50:18.407 [INFO] ZooKeeperQueueTest.run(80) 消费对象: 10 , 然后等待 2400 毫秒

4      队列的消费者

消费者尝试从子节点列表获取zNode名最小的一个子节点，如果队列为空则等待NodeChildrenChanged事件。关键代码

/** 队列的同步信号 */

private static Integer queueMutex = Integer.valueOf(1);

@Override

protected void processNodeChildrenChanged(WatchedEvent event) {

synchronized (queueMutex) {

queueMutex.notify();

}

}

/**

* 从队列中删除第一个对象

*

* @return

* @throws KeeperException

* @throws InterruptedException

*/

int consume() throws KeeperException, InterruptedException {

while (true) {

synchronized (queueMutex) {

List<String> list = getZooKeeper().getChildren(queueName, true);

if (list.size() == 0) {

queueMutex.wait();

} else {

// 获取第一个子节点的名称

String firstNodeName = getFirstElementName(list);

// 删除节点，并返回节点的值

return deleteNodeAndReturnValue(firstNodeName);

}

}

}

}

5      完整源码

5.1   ZooKeeperBase.java

package tech.codestory.zookeeper;

import java.io.IOException;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.*;

import org.apache.zookeeper.data.Stat;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.slf4j.profiler.Profiler;

/**

* 为 ZooKeeper测试代码创建一个基类，封装建立连接的过程

*

* @author code story

* @date 2019/8/16

*/

public class ZooKeeperBase implements Watcher {

/** 日志，不使用 @Slf4j ，是要使用子类的log */

Logger log = null;

/** 等待连接建立成功的信号 */

private CountDownLatch connectedSemaphore = new CountDownLatch(1);

/** ZooKeeper 客户端 */

private ZooKeeper zooKeeper = null;

/** 避免重复根节点 */

static Integer rootNodeInitial = Integer.valueOf(1);

/** 构造函数 */

public ZooKeeperBase(String address) throws IOException {

log = LoggerFactory.getLogger(getClass());

Profiler profiler = new Profiler(this.getClass().getName() + " 连接到ZooKeeper");

profiler.start("开始链接");

zooKeeper = new ZooKeeper(address, 3000, this);

try {

profiler.start("等待连接成功的Event");

connectedSemaphore.await();

} catch (InterruptedException e) {

log.error("InterruptedException", e);

}

profiler.stop();

profiler.setLogger(log);

profiler.log();

}

/**

* 创建测试需要的根节点

*

* @param rootNodeName

* @return

*/

public String createRootNode(String rootNodeName) {

synchronized (rootNodeInitial) {

// 创建 tableSerial 的zNode

try {

Stat existsStat = getZooKeeper().exists(rootNodeName, false);

if (existsStat == null) {

rootNodeName = getZooKeeper().create(rootNodeName, new byte[0],

ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

}

} catch (KeeperException e) {

log.error("KeeperException", e);

} catch (InterruptedException e) {

log.error("InterruptedException", e);

}

}

return rootNodeName;

}

/** 读取ZooKeeper对象，供子类调用 */

protected ZooKeeper getZooKeeper() {

return zooKeeper;

}

@Override

final public void process(WatchedEvent event) {

if (Event.EventType.None.equals(event.getType())) {

// 连接状态发生变化

if (Event.KeeperState.SyncConnected.equals(event.getState())) {

// 连接建立成功

connectedSemaphore.countDown();

}

} else if (Event.EventType.NodeCreated.equals(event.getType())) {

processNodeCreated(event);

} else if (Event.EventType.NodeDeleted.equals(event.getType())) {

processNodeDeleted(event);

} else if (Event.EventType.NodeDataChanged.equals(event.getType())) {

processNodeDataChanged(event);

} else if (Event.EventType.NodeChildrenChanged.equals(event.getType())) {

processNodeChildrenChanged(event);

}

}

/**

* 处理事件: NodeCreated

*

* @param event

*/

protected void processNodeCreated(WatchedEvent event) {}

/**

* 处理事件: NodeDeleted

*

* @param event

*/

protected void processNodeDeleted(WatchedEvent event) {}

/**

* 处理事件: NodeDataChanged

*

* @param event

*/

protected void processNodeDataChanged(WatchedEvent event) {}

/**

* 处理事件: NodeChildrenChanged

*

* @param event

*/

protected void processNodeChildrenChanged(WatchedEvent event) {}

}

5.2   ZooKeeperQueue.java

package tech.codestory.zookeeper.queue;

import java.io.IOException;

import java.nio.ByteBuffer;

import java.util.ArrayList;

import java.util.List;

import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.KeeperException;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.ZooDefs;

import org.apache.zookeeper.data.ACL;

import org.apache.zookeeper.data.Stat;

import lombok.extern.slf4j.Slf4j;

import tech.codestory.zookeeper.ZooKeeperBase;

/**

* ZooKeeper实现Queue

*

* @author code story

* @date 2019/8/16

*/

@Slf4j

public class ZooKeeperQueue extends ZooKeeperBase {

/** 队列名称 */

private String queueName;

/** 队列的同步信号 */

private static Integer queueMutex = Integer.valueOf(1);

/**

* 构造函数

*

* @param address

* @param queueName

* @throws IOException

*/

public ZooKeeperQueue(String address, String queueName) throws IOException {

super(address);

this.queueName = createRootNode(queueName);

}

@Override

protected void processNodeChildrenChanged(WatchedEvent event) {

synchronized (queueMutex) {

queueMutex.notify();

}

}

/**

* 将对象添加到队列中

*

* @param i

* @return

*/

boolean produce(int i) throws KeeperException, InterruptedException {

ByteBuffer b = ByteBuffer.allocate(4);

byte[] value;

// Add child with value i

b.putInt(i);

value = b.array();

String elementName = queueName + "/element";

ArrayList<ACL> ids = ZooDefs.Ids.OPEN_ACL_UNSAFE;

CreateMode createMode = CreateMode.PERSISTENT_SEQUENTIAL;

getZooKeeper().create(elementName, value, ids, createMode);

return true;

}

/**

* 从队列中删除第一个对象

*

* @return

* @throws KeeperException

* @throws InterruptedException

*/

int consume() throws KeeperException, InterruptedException {

while (true) {

synchronized (queueMutex) {

List<String> list = getZooKeeper().getChildren(queueName, true);

if (list.size() == 0) {

queueMutex.wait();

} else {

// 获取第一个子节点的名称

String firstNodeName = getFirstElementName(list);

// 删除节点，并返回节点的值

return deleteNodeAndReturnValue(firstNodeName);

}

}

}

}

/**

* 获取第一个子节点的名称

*

* @param list

* @return

*/

private String getFirstElementName(List<String> list) {

Integer min = Integer.MAX_VALUE;

String minNode = null;

for (String s : list) {

Integer tempValue = Integer.valueOf(s.substring(7));

if (tempValue < min) {

min = tempValue;

minNode = s;

}

}

return minNode;

}

/**

* 删除节点，并返回节点的值

*

* @param minNode

* @return

* @throws KeeperException

* @throws InterruptedException

*/

private int deleteNodeAndReturnValue(String minNode)

throws KeeperException, InterruptedException {

String fullNodeName = queueName + "/" + minNode;

Stat stat = new Stat();

byte[] b = getZooKeeper().getData(fullNodeName, false, stat);

getZooKeeper().delete(fullNodeName, stat.getVersion());

ByteBuffer buffer = ByteBuffer.wrap(b);

return buffer.getInt();

}

}

5.3   ZooKeeperQueueTest.java

package tech.codestory.zookeeper.queue;

import java.io.IOException;

import java.security.SecureRandom;

import java.util.Random;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.KeeperException;

import org.testng.annotations.Test;

import lombok.extern.slf4j.Slf4j;

/**

* ZooKeeperQueue测试

*

* @author code story

* @date 2019/8/16

*/

@Slf4j

public class ZooKeeperQueueTest {

final String address = "192.168.5.128:2181";

final String queueName = "/queue";

final Random random = new SecureRandom();

// 随机生成10-20之间的个数

final int count = 10 + random.nextInt(10);

/** 等待生产者和消费者线程都结束 */

private CountDownLatch connectedSemaphore = new CountDownLatch(2);

@Test

public void testQueue() {

log.info("开始ZooKeeper队列测试，本次将测试 {} 个数据", count);

new QueueProducer().start();

new QueueConsumer().start();

try {

connectedSemaphore.await();

} catch (InterruptedException e) {

log.error("InterruptedException", e);

}

}

/**

* 队列的生产者

*/

class QueueProducer extends Thread {

@Override

public void run() {

try {

ZooKeeperQueue queue = new ZooKeeperQueue(address, queueName);

for (int i = 0; i < count; i++) {

int elementValue = i + 1;

long waitTime = random.nextInt(50) * 100;

log.info("生产对象 : {} , 然后等待 {} 毫秒", elementValue, waitTime);

queue.produce(elementValue);

Thread.sleep(waitTime);

}

} catch (IOException e) {

log.error("IOException", e);

} catch (InterruptedException e) {

log.error("InterruptedException", e);

} catch (KeeperException e) {

log.error("KeeperException", e);

}

connectedSemaphore.countDown();

}

}

/**

* 队列的消费者

*/

class QueueConsumer extends Thread {

@Override

public void run() {

try {

ZooKeeperQueue queue = new ZooKeeperQueue(address, queueName);

for (int i = 0; i < count; i++) {

try {

int elementValue = queue.consume();

long waitTime = random.nextInt(50) * 100;

log.info("消费对象: {} , 然后等待 {} 毫秒", elementValue, waitTime);

Thread.sleep(waitTime);

} catch (KeeperException e) {

i--;

log.error("KeeperException", e);

} catch (InterruptedException e) {

log.error("InterruptedException", e);

}

}

connectedSemaphore.countDown();

} catch (IOException e) {

log.error("IOException", e);

}

}

}

}

最新文章

  1. BI系统与KPI指标的整合分析
  2. 使用opencv设置图像的格式以及帧率
  3. cordova 开发属于自己的插件---android
  4. PSP记录表
  5. HDU4725 The Shortest Path in Nya Graph SPFA最短路
  6. HDU-1391 Number Steps
  7. Android学习笔记（一）Android应用程序的组成部分
  8. java_IO流之 NIO
  9. C/C++中慎用宏(#define)
  10. 使用RandomAccessFile类对文件进行读写
  11. .net接收post请求，并转为字符串
  12. vue上线后，背景图片路径错误
  13. [控件] LineAnimationView
  14. php设计模式-工厂设计模式
  15. eclipse maven 报错Could not get the value for parameter encoding for plugin execution default
  16. Hessain 方法重载
  17. python学习笔记（日志系统实现）
  18. VS 正则表达式替换内容
  19. IP Fragmentation(IP分片)
  20. Trident学习笔记（二）

热门文章

  1. VC win32 static library静态链接库简单示例
  2. [leetcode] 8. String to Integer (atoi) (Medium)
  3. [HDOJ] 1753.大明A+B （大数加法）
  4. linux初学者-系统日志（二）
  5. spring-boot-plus后台快速开发脚手架之代码生成器使用
  6. 10w数组去重，排序，找最多出现次数（精华）
  7. python 爬取豆瓣电影评论，并进行词云展示及出现的问题解决办法
  8. Ubuntu 16.04 LTS设置屏幕分辨率并永久保存所设置的分辨率
  9. 【iOS】删除 main.storyboard 的问题
  10. 关于定时器Scheduled(cron)的问题