【欢迎关注公众号：程序猿讲故事 （codestory），及时接收最新文章】
生产-消费者队列，用于多节点的分布式数据结构，生产和消费数据。生产者创建一个数据对象，并放到队列中；消费者从队列中取出一个数据对象并进行处理。在zookeeper中，队列可以使用一个容器节点下创建多个子节点来实现；创建子节点时，createmode使用 persistent_sequential，zookeeper会自动在节点名称后面添加唯一序列号。ephemeral_sequential也有同样的特点，区别在于会话结束后是否会自动删除。
敲小黑板：*_sequential是zookeeper的一个很重要的特性，分布式锁、选举制度都依靠这个特性实现的。
之前的文章，我们已经用实现了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(); } } |
生产者的关键代码
|
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。
消费者尝试从子节点列表获取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); } } } } |
把测试结果放源码前面，免得大家被无聊的代码晃晕。
测试代码创建了两个线程，一个线程是生产者，按随机间隔往队列中添加对象；一个线程是消费者，随机间隔尝试从队列中取出第一个，如果当时队列为空，会等到直到新的数据。
两个进程都加上随机间隔，是为了模拟生产可能比消费更快的情况。以下是测试日志，为了更突出，生产和消费的日志我增加了不同的文字样式。
|
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 毫秒 |
|
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) {} } |
|
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(); } } |
|
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); } } } } |
如对本文有疑问， 点击进行留言回复！！
如何pick分布式系统协调框架？Eureka or Zookeeper？
我要评论
网友评论