Zookeeper分布式队列的实现

浅浅的花香味﹌ 2021-09-12 06:00 170阅读 0赞

摘要：本文要通过zookeeper实现一个简单可靠的分布式队列

本文源码请在这里下载：[https://github.com/appleappleapple/DistributeLearning][https_github.com_appleappleapple_DistributeLearning]

一、队列

Zookeeper可以处理两种类型的队列：  
（1）同步队列  
当一个队列的成员都聚齐时，这个队列才可用，否则一直等待所有成员到达。例如一个班去旅游，看是否所有人都到齐了，到齐了就发车。例如有个大任务分解为多个子任务，要所有子任务都完成了才能进入到下一流程。  
（2）先进先出队列  
按照FIFO方式进行入队和出队

例如实现生产者和消费者模型

二、实现思路  
（1）同步队列  
在zookeeper中先创建一个根目录 queue\_sync，做为队列队列的消费者监视/queue/start节点，刚开始还没有这个节点，所以什么都不会做。入队操作就是在queue\_sync下创建子节点，然后计算子节点的总数，看是否和队列的目标数量相同。如果相同，创建/queue\_sync/start节点，由于/queue\_sync/start这个节点有了状态变化，zookeeper就会通知监视者：队员已经到齐了，监视者得到通知后进行自己的后续流程

实现代码

package com.github.distribute.queue;
    
    import java.io.ByteArrayInputStream;
    import java.io.ByteArrayOutputStream;
    import java.io.ObjectInputStream;
    import java.io.ObjectOutputStream;
    import java.util.Collections;
    import java.util.Comparator;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    import org.I0Itec.zkclient.ExceptionUtil;
    import org.I0Itec.zkclient.exception.ZkNoNodeException;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.EventType;
    import org.apache.zookeeper.ZooDefs;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.data.Stat;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * 分布式队列，同步队列的实现
     * 
     * @author linbingwen
     *
     * @param <T>
     */
    public class DistributedQueue<T> {
    	private static Logger logger = LoggerFactory.getLogger(DistributedQueue.class);
    
    	protected final ZooKeeper zooKeeper;// 用于操作zookeeper集群
    	protected final String root;// 代表根节点
    	private int queueSize;
    	private String startPath = "/queue/start";
    
    	protected static final String Node_NAME = "n_";// 顺序节点的名称
    
    	public DistributedQueue(ZooKeeper zooKeeper, String root, int queueSize) {
    		this.zooKeeper = zooKeeper;
    		this.root = root;
    		this.queueSize = queueSize;
    		init();
    	}
    
    	/**
    	 * 初始化根目录
    	 */
    	private void init() {
    		try {
    			Stat stat = zooKeeper.exists(root, false);// 判断一下根目录是否存在
    			if (stat == null) {
    				zooKeeper.create(root, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    			}
    			zooKeeper.delete(startPath, -1); // 删除队列满的标志
    		} catch (Exception e) {
    			logger.error("create rootPath error", e);
    		}
    	}
    
    	/**
    	 * 获取队列的大小
    	 * 
    	 * @return
    	 * @throws Exception
    	 */
    	public int size() throws Exception {
    		return zooKeeper.getChildren(root, false).size();
    	}
    
    	/**
    	 * 判断队列是否为空
    	 * 
    	 * @return
    	 * @throws Exception
    	 */
    	public boolean isEmpty() throws Exception {
    		return zooKeeper.getChildren(root, false).size() == 0;
    	}
    
    	/**
    	 * bytes 转object
    	 * 
    	 * @param bytes
    	 * @return
    	 */
    	private Object ByteToObject(byte[] bytes) {
    		Object obj = null;
    		try {
    			// bytearray to object
    			ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
    			ObjectInputStream oi = new ObjectInputStream(bi);
    
    			obj = oi.readObject();
    			bi.close();
    			oi.close();
    		} catch (Exception e) {
    			logger.error("translation" + e.getMessage());
    			e.printStackTrace();
    		}
    		return obj;
    	}
    
    	/**
    	 * Object 转byte
    	 * 
    	 * @param obj
    	 * @return
    	 */
    	private byte[] ObjectToByte(java.lang.Object obj) {
    		byte[] bytes = null;
    		try {
    			// object to bytearray
    			ByteArrayOutputStream bo = new ByteArrayOutputStream();
    			ObjectOutputStream oo = new ObjectOutputStream(bo);
    			oo.writeObject(obj);
    
    			bytes = bo.toByteArray();
    
    			bo.close();
    			oo.close();
    		} catch (Exception e) {
    			logger.error("translation" + e.getMessage());
    			e.printStackTrace();
    		}
    		return bytes;
    	}
    
    	/**
    	 * 向队列提供数据,队列满的话会阻塞等待直到start标志位清除
    	 * 
    	 * @param element
    	 * @return
    	 * @throws Exception
    	 */
    	public boolean offer(T element) throws Exception {
    		// 构建数据节点的完整路径
    		String nodeFullPath = root.concat("/").concat(Node_NAME);
    		try {
    			if (queueSize > size()) {
    				// 创建持久的节点，写入数据
    				zooKeeper.create(nodeFullPath, ObjectToByte(element), ZooDefs.Ids.OPEN_ACL_UNSAFE,
    						CreateMode.PERSISTENT);
    				// 再判断一下队列是否满
    				if (queueSize > size()) {
    					zooKeeper.delete(startPath, -1); // 确保不存在
    				} else {
    					zooKeeper.create(startPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    				}
    			} else {
    				// 创建队列满的标记
    				if (zooKeeper.exists(startPath, false) != null) {
    					zooKeeper.create(startPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    				}
    
    				final CountDownLatch latch = new CountDownLatch(1);
    				final Watcher previousListener = new Watcher() {
    					public void process(WatchedEvent event) {
    						if (event.getType() == EventType.NodeDeleted) {
    							latch.countDown();
    						}
    					}
    				};
    
    				// 如果节点不存在会出现异常
    				zooKeeper.exists(startPath, previousListener);
    				latch.await();
    				offer(element);
    
    			}
    		} catch (ZkNoNodeException e) {
    			logger.error("", e);
    		} catch (Exception e) {
    			throw ExceptionUtil.convertToRuntimeException(e);
    		}
    		return true;
    	}
    
    	/**
    	 * 从队列取数据,当有start标志位时，开始取数据，全部取完数据后才删除start标志
    	 * 
    	 * @return
    	 * @throws Exception
    	 */
    	@SuppressWarnings("unchecked")
    	public T poll() throws Exception {
    
    		try {
    			// 队列还没满
    			if (zooKeeper.exists(startPath, false) == null) {
    				final CountDownLatch latch = new CountDownLatch(1);
    				final Watcher previousListener = new Watcher() {
    					public void process(WatchedEvent event) {
    						if (event.getType() == EventType.NodeCreated) {
    							latch.countDown();
    						}
    					}
    				};
    
    				// 如果节点不存在会出现异常
    				zooKeeper.exists(startPath, previousListener);
    
    				// 如果节点不存在会出现异常
    				latch.await();
    			}
    
    			List<String> list = zooKeeper.getChildren(root, false);
    			if (list.size() == 0) {
    				return null;
    			}
    			// 将队列按照由小到大的顺序排序
    			Collections.sort(list, new Comparator<String>() {
    				public int compare(String lhs, String rhs) {
    					return getNodeNumber(lhs, Node_NAME).compareTo(getNodeNumber(rhs, Node_NAME));
    				}
    			});
    
    			/**
    			 * 将队列中的元素做循环，然后构建完整的路径，在通过这个路径去读取数据
    			 */
    			for (String nodeName : list) {
    				String nodeFullPath = root.concat("/").concat(nodeName);
    				try {
    					T node = (T) ByteToObject(zooKeeper.getData(nodeFullPath, false, null));
    					zooKeeper.delete(nodeFullPath, -1);
    					return node;
    				} catch (ZkNoNodeException e) {
    					logger.error("", e);
    				}
    			}
    			return null;
    		} catch (Exception e) {
    			throw ExceptionUtil.convertToRuntimeException(e);
    		}
    
    	}
    
    	/**
    	 * 截取节点的数字的方法
    	 * 
    	 * @param str
    	 * @param nodeName
    	 * @return
    	 */
    	private String getNodeNumber(String str, String nodeName) {
    		int index = str.lastIndexOf(nodeName);
    		if (index >= 0) {
    			index += Node_NAME.length();
    			return index <= str.length() ? str.substring(index) : "";
    		}
    		return str;
    
    	}
    
    }

**代码还没验证，可能会有问题！**

（2）先进先出队列

在zookeeper中先创建一个根目录 queue\_fifo，做为队列。入队操作就是在queue\_fifo下创建自增序的子节点，并把数据放入节点内。出队操作就是先找到queue\_fifo下序号最下的那个节点，取出数据，然后删除此节点。

实现代码：

package com.github.distribute.queue;
    
    import java.util.Collections;
    import java.util.Comparator;
    import java.util.List;
    import org.I0Itec.zkclient.ExceptionUtil;
    import org.I0Itec.zkclient.ZkClient;
    import org.I0Itec.zkclient.exception.ZkNoNodeException;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.github.distribute.lock.zookeeper.BaseDistributedLock;
    
    /**
     * 分布式队列，生产者，消费者的实现
     * @author linbingwen
     *
     * @param <T>
     */
    public class DistributedSimpleQueue<T> {
    
    	private static Logger logger = LoggerFactory.getLogger(BaseDistributedLock.class);
    
    	protected final ZkClient zkClient;//用于操作zookeeper集群
    	protected final String root;//代表根节点
    
    	protected static final String Node_NAME = "n_";//顺序节点的名称
    	
    
    
    	public DistributedSimpleQueue(ZkClient zkClient, String root) {
    		this.zkClient = zkClient;
    		this.root = root;
    	}
        
    	//获取队列的大小
    	public int size() {
    		/**
    		 * 通过获取根节点下的子节点列表
    		 */
    		return zkClient.getChildren(root).size();
    	}
    	
        //判断队列是否为空
    	public boolean isEmpty() {
    		return zkClient.getChildren(root).size() == 0;
    	}
    	
    	/**
    	 * 向队列提供数据
    	 * @param element
    	 * @return
    	 * @throws Exception
    	 */
        public boolean offer(T element) throws Exception{
        	
        	//构建数据节点的完整路径
        	String nodeFullPath = root .concat( "/" ).concat( Node_NAME );
            try {
            	//创建持久的节点，写入数据
                zkClient.createPersistentSequential(nodeFullPath , element);
            }catch (ZkNoNodeException e) {
            	zkClient.createPersistent(root);
            	offer(element);
            } catch (Exception e) {
                throw ExceptionUtil.convertToRuntimeException(e);
            }
            return true;
        }
    
    
        //从队列取数据
    	@SuppressWarnings("unchecked")
    	public T poll() throws Exception {
    		
    		try {
    
    			List<String> list = zkClient.getChildren(root);
    			if (list.size() == 0) {
    				return null;
    			}
    			//将队列按照由小到大的顺序排序
    			Collections.sort(list, new Comparator<String>() {
    				public int compare(String lhs, String rhs) {
    					return getNodeNumber(lhs, Node_NAME).compareTo(getNodeNumber(rhs, Node_NAME));
    				}
    			});
    			
    			/**
    			 * 将队列中的元素做循环，然后构建完整的路径，在通过这个路径去读取数据
    			 */
    			for ( String nodeName : list ){
    				
    				String nodeFullPath = root.concat("/").concat(nodeName);	
    				try {
    					T node = (T) zkClient.readData(nodeFullPath);
    					zkClient.delete(nodeFullPath);
    					return node;
    				} catch (ZkNoNodeException e) {
    					logger.error("",e);
    				}
    			}
    			
    			return null;
    			
    		} catch (Exception e) {
    			throw ExceptionUtil.convertToRuntimeException(e);
    		}
    
    	}
    
    	
    	private String getNodeNumber(String str, String nodeName) {
    		int index = str.lastIndexOf(nodeName);
    		if (index >= 0) {
    			index += Node_NAME.length();
    			return index <= str.length() ? str.substring(index) : "";
    		}
    		return str;
    
    	}
    
    }

测试一下：

package com.github.distribute.queue;
    
    import java.io.Serializable;
    
    import org.I0Itec.zkclient.ZkClient;
    import org.I0Itec.zkclient.serialize.SerializableSerializer;
    
    public class DistributedQueueTest {
    
    	public static void main(String[] args) {
    		ZkClient zkClient = new ZkClient("127.0.0.1:2181", 5000, 5000, new SerializableSerializer());
    		DistributedSimpleQueue<SendObject> queue = new DistributedSimpleQueue<SendObject>(zkClient, "/Queue");
    		new Thread(new ConsumerThread(queue)).start();
    		new Thread(new ProducerThread(queue)).start();
    
    	}
    
    }
    
    class ConsumerThread implements Runnable {
    	private DistributedSimpleQueue<SendObject> queue;
    
    	public ConsumerThread(DistributedSimpleQueue<SendObject> queue) {
    		this.queue = queue;
    	}
    
    	public void run() {
    		for (int i = 0; i < 10000; i++) {
    			try {
    				Thread.sleep((int) (Math.random() * 5000));// 随机睡眠一下
    				SendObject sendObject = (SendObject) queue.poll();
    				System.out.println("消费一条消息成功：" + sendObject);
    			} catch (Exception e) {
    			}
    		}
    	}
    }
    
    class ProducerThread implements Runnable {
    
    	private DistributedSimpleQueue<SendObject> queue;
    
    	public ProducerThread(DistributedSimpleQueue<SendObject> queue) {
    		this.queue = queue;
    	}
    
    	public void run() {
    		for (int i = 0; i < 10000; i++) {
    			try {
    				Thread.sleep((int) (Math.random() * 5000));// 随机睡眠一下
    				SendObject sendObject = new SendObject(String.valueOf(i), "content" + i);
    				queue.offer(sendObject);
    				System.out.println("发送一条消息成功：" + sendObject);
    			} catch (Exception e) {
    			}
    		}
    	}
    
    }
    
    class SendObject implements Serializable {
    
    	private static final long serialVersionUID = 1L;
    
    	public SendObject(String id, String content) {
    		this.id = id;
    		this.content = content;
    	}
    
    	private String id;
    
    	private String content;
    
    	public String getId() {
    		return id;
    	}
    
    	public void setId(String id) {
    		this.id = id;
    	}
    
    	public String getContent() {
    		return content;
    	}
    
    	public void setContent(String content) {
    		this.content = content;
    	}
    
    	@Override
    	public String toString() {
    		return "SendObject [id=" + id + ", content=" + content + "]";
    	}
    
    }

输出结果：

![20170426170923858][]

本文源码请在这里下载：[https://github.com/appleappleapple/DistributeLearning][https_github.com_appleappleapple_DistributeLearning]

更多技术请关注笔者微信技术公众号"单例模式"

![70][]

[https_github.com_appleappleapple_DistributeLearning]: https://github.com/appleappleapple/DistributeLearning
[20170426170923858]: /images/20210811/8680cb3fbfaf4955a21a72801f17c216.png
[70]: /images/20210811/84aa18eed235456d88af1e6a2aa6f27b.png