基于ZooKeeper实现队列源码

2024-07-13 10:11:40

字体：大中小

来源：转载

供稿：网友

实现原理

先进先出队列是最常用的队列，使用Zookeeper实现先进先出队列就是在特定的目录下创建PERSISTENT_EQUENTIAL节点，创建成功时Watcher通知等待的队列，队列删除序列号最小的节点用以消费。此场景下Zookeeper的znode用于消息存储，znode存储的数据就是消息队列中的消息内容，SEQUENTIAL序列号就是消息的编号，按序取出即可。由于创建的节点是持久化的，所以不必担心队列消息的丢失问题。

队列（Queue）

分布式队列是通用的数据结构，为了在 Zookeeper 中实现分布式队列，首先需要指定一个 Znode 节点作为队列节点（queue node），各个分布式客户端通过调用 create() 函数向队列中放入数据，调用create()时节点路径名带"qn-"结尾，并设置顺序（sequence）节点标志。由于设置了节点的顺序标志，新的路径名具有以下字符串模式："_path-to-queue-node_/qn-X"，X 是唯一自增号。需要从队列中获取数据/移除数据的客户端首先调用 getChildren() 函数，有数据则获取（获取数据后可以删除也可以不删），没有则在队列节点（queue node）上将 watch 设置为 true，等待触发并处理最小序号的节点（即从序号最小的节点中取数据）。

应用场景

Zookeeper队列不太适合要求高性能的场合，但可以在数据量不大的情况下考虑使用。比如已在项目中使用Zookeeper又需要小规模的队列应用，这时可以使用Zookeeper实现的队列;毕竟引进一个消息中间件会增加系统的复杂性和运维的压力。

详细代码

ZookeeperClient工具类

java;">package org.massive.common; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import java.io.IOException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; /**  * Created by Massive on 2016/12/18.  */ public class ZooKeeperClient {  private static String connectionString = "localhost:2181";  private static int sessionTimeout = 10000;  public static ZooKeeper getInstance() throws IOException, InterruptedException {  //--------------------------------------------------------------  // 为避免连接还未完成就执行zookeeper的get/create/exists操作引起的（KeeperErrorCode = ConnectionLoss)  // 这里等Zookeeper的连接完成才返回实例  //--------------------------------------------------------------  final CountDownLatch connectedSignal = new CountDownLatch(1);  ZooKeeper zk = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {   @Override   public void process(WatchedEvent event) {    if (event.getState() == Event.KeeperState.SyncConnected) {    connectedSignal.countDown();    } else if (event.getState() == Event.KeeperState.Expired) {    }   }   });  connectedSignal.await(sessionTimeout, TimeUnit.MILLISECONDS);  return zk;  }  public static int getSessionTimeout() {  return sessionTimeout;  }  public static void setSessionTimeout(int sessionTimeout) {  ZooKeeperClient.sessionTimeout = sessionTimeout;  } }

ZooKeeperQueue

package org.massive.queue; import org.apache.commons.lang3.RandomUtils; import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat; import org.massive.common.ZooKeeperClient; import java.io.IOException; import java.io.UnsupportedEncodingException; import java.util.List; import java.util.SortedSet; import java.util.TreeSet; /**  * Created by Allen on 2016/12/22.  */ public class ZooKeeperQueue {  private ZooKeeper zk;  private int sessionTimeout;  private static byte[] ROOT_QUEUE_DATA = {0x12,0x34};  private static String QUEUE_ROOT = "/QUEUE";  private String queueName;  private String queuePath;  private Object mutex = new Object();  public ZooKeeperQueue(String queueName) throws IOException, KeeperException, InterruptedException {  this.queueName = queueName;  this.queuePath = QUEUE_ROOT + "/" + queueName;  this.zk = ZooKeeperClient.getInstance();  this.sessionTimeout = zk.getSessionTimeout();  //----------------------------------------------------  // 确保队列根目录/QUEUE和当前队列的目录的存在  //----------------------------------------------------  ensureExists(QUEUE_ROOT);  ensureExists(queuePath);  }  public byte[] consume() throws InterruptedException, KeeperException, UnsupportedEncodingException {  List<String> nodes = null;  byte[] returnVal = null;  Stat stat = null;  do {   synchronized (mutex) {   nodes = zk.getChildren(queuePath, new ProduceWatcher());   //----------------------------------------------------   // 如果没有消息节点，等待生产者的通知   //----------------------------------------------------   if (nodes == null || nodes.size() == 0) {    mutex.wait();   } else {    SortedSet<String> sortedNode = new TreeSet<String>();    for (String node : nodes) {    sortedNode.add(queuePath + "/" + node);    }    //----------------------------------------------------    // 消费队列里序列号最小的消息    //----------------------------------------------------    String first = sortedNode.first();    returnVal = zk.getData(first, false, stat);    zk.delete(first, -1);    System.out.print(Thread.currentThread().getName() + " ");    System.out.print("consume a message from queue：" + first);    System.out.println(", message data is: " + new String(returnVal,"UTF-8"));    return returnVal;   }   }  } while (true);  }  class ProduceWatcher implements Watcher {  @Override  public void process(WatchedEvent event) {   //----------------------------------------------------   // 生产一条消息成功后通知一个等待线程   //----------------------------------------------------   synchronized (mutex) {   mutex.notify();   }  }  }  public void produce(byte[] data) throws KeeperException, InterruptedException, UnsupportedEncodingException {  //----------------------------------------------------  // 确保当前队列目录存在  // example: /QUEUE/queueName  //----------------------------------------------------  ensureExists(queuePath);  String node = zk.create(queuePath + "/", data,   ZooDefs.Ids.OPEN_ACL_UNSAFE,   CreateMode.PERSISTENT_SEQUENTIAL);  System.out.print(Thread.currentThread().getName() + " ");  System.out.print("produce a message to queue：" + node);  System.out.println(" , message data is: " + new String(data,"UTF-8"));  }  public void ensureExists(String path) {  try {   Stat stat = zk.exists(path, false);   if (stat == null) {   zk.create(path, ROOT_QUEUE_DATA, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);   }  } catch (KeeperException e) {   e.printStackTrace();  } catch (InterruptedException e) {   e.printStackTrace();  }  }  public static void main(String[] args) throws IOException, InterruptedException, KeeperException {  String queueName = "test";  final ZooKeeperQueue queue = new ZooKeeperQueue(queueName);  for (int i = 0; i < 10; i++) {   new Thread(new Runnable() {   @Override   public void run() {    try {    queue.consume();    System.out.println("--------------------------------------------------------");    System.out.println();    } catch (InterruptedException e) {    e.printStackTrace();    } catch (KeeperException e) {    e.printStackTrace();    } catch (UnsupportedEncodingException e) {    e.printStackTrace();    }   }   }).start();  }  new Thread(new Runnable() {   @Override   public void run() {   for (int i = 0; i < 10; i++) {    try {    Thread.sleep(RandomUtils.nextInt(100 * i, 200 * i));    queue.produce(("massive" + i).getBytes());    } catch (InterruptedException e) {    e.printStackTrace();    } catch (KeeperException e) {    e.printStackTrace();    } catch (UnsupportedEncodingException e) {    e.printStackTrace();    }   }   }  },"Produce-thread").start();  } }

测试

运行main方法，本机器的某次输出结果

Produce-thread produce a message to queue：/QUEUE/test/0000000000 , message data is: massive0 Thread-8 consume a message from queue：/QUEUE/test/0000000000, message data is: massive0 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000001 , message data is: massive1 Thread-6 consume a message from queue：/QUEUE/test/0000000001, message data is: massive1 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000002 , message data is: massive2 Thread-3 consume a message from queue：/QUEUE/test/0000000002, message data is: massive2 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000003 , message data is: massive3 Thread-0 consume a message from queue：/QUEUE/test/0000000003, message data is: massive3 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000004 , message data is: massive4 Thread-5 consume a message from queue：/QUEUE/test/0000000004, message data is: massive4 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000005 , message data is: massive5 Thread-2 consume a message from queue：/QUEUE/test/0000000005, message data is: massive5 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000006 , message data is: massive6 Thread-4 consume a message from queue：/QUEUE/test/0000000006, message data is: massive6 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000007 , message data is: massive7 Thread-9 consume a message from queue：/QUEUE/test/0000000007, message data is: massive7 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000008 , message data is: massive8 Thread-7 consume a message from queue：/QUEUE/test/0000000008, message data is: massive8 -------------------------------------------------------- Produce-thread produce a message to queue：/QUEUE/test/0000000009 , message data is: massive9 Thread-1 consume a message from queue：/QUEUE/test/0000000009, message data is: massive9

总结

以上就是本文有关于队列和基于ZooKeeper实现队列源码介绍的全部内容，希望对大家有所帮助。

感谢朋友们对本站的支持！

注：相关教程知识阅读请移步到JAVA教程频道。

上一篇：Default Methods实例解析

下一篇：Java正则替换手机号代码实例