博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
Zookeeper(八)分布式队列
阅读量:6241 次
发布时间:2019-06-22

本文共 5951 字,大约阅读时间需要 19 分钟。

1. element  方法  获取对列头部第一个元素

 查找队列znode 下所有的子节点名称   使用TreeMap给顺序编号排序  返回第一个znode对应的值

 

public byte[] element() throws NoSuchElementException, KeeperException, InterruptedException {        TreeMap
orderedChildren; while(true){ try{ // 返回对列中的全部元素name 这里使用到了TreeMap // map的key为znode的顺序编号 orderedChildren = orderedChildren(null); }catch(KeeperException.NoNodeException e){ throw new NoSuchElementException(); } if(orderedChildren.size() == 0 ) throw new NoSuchElementException(); for(String headNode : orderedChildren.values()){ if(headNode != null){ try{ // 返回对列头部第一个znode对应数据. return zookeeper.getData(dir+"/"+headNode, false, null); }catch(KeeperException.NoNodeException e){ } } } } }

获取所有子节点名称

 

 

private TreeMap
orderedChildren(Watcher watcher) throws KeeperException, InterruptedException { TreeMap
orderedChildren = new TreeMap
(); List
childNames = null; try{ // 返回对列节点下 所有的子znode 名称 childNames = zookeeper.getChildren(dir, watcher); }catch (KeeperException.NoNodeException e){ throw e; } // 所有子节点 for(String childName : childNames){ try{ //Check format if(!childName.regionMatches(0, prefix, 0, prefix.length())){ LOG.warn("Found child node with improper name: " + childName); continue; } String suffix = childName.substring(prefix.length()); // 顺序节点编号 Long childId = new Long(suffix); orderedChildren.put(childId,childName); }catch(NumberFormatException e){ LOG.warn("Found child node with improper format : " + childName + " " + e,e); } } return orderedChildren; }

 

2.   remove  返回对列头部的第一个元素的值  并删除该znode

 

public byte[] remove() throws NoSuchElementException, KeeperException, InterruptedException {        TreeMap
orderedChildren; while(true){ try{ // 查找所有子节点 用TreeMap排序 orderedChildren = orderedChildren(null); }catch(KeeperException.NoNodeException e){ throw new NoSuchElementException(); } if(orderedChildren.size() == 0) throw new NoSuchElementException(); for(String headNode : orderedChildren.values()){ String path = dir +"/"+headNode; try{ // 对列头部 第一个节点对应的数据 byte[] data = zookeeper.getData(path, false, null); // 删除该节点 zookeeper.delete(path, -1); return data; }catch(KeeperException.NoNodeException e){ // Another client deleted the node first. } } } }

3. take 检索并移除对列头部一个znode对应的值  如果对列为空  则一直等待

 

 

public byte[] take() throws KeeperException, InterruptedException {        TreeMap
orderedChildren; // Same as for element. Should refactor this. while(true){ LatchChildWatcher childWatcher = new LatchChildWatcher(); try{ // 查找所有对列元素 并给对列主znode 设置监视器 orderedChildren = orderedChildren(childWatcher); }catch(KeeperException.NoNodeException e){ zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT); continue; } // 如果对列中不存在元素 // 对列节点下不存在子节点, 线程将一直等待. 使用到CountDownLatch // 当客户端调用 offer方法 写入一个元素时 触发LatchChildWatcher监视器CountDownLatch 计数减1 为0时 当前线程获得运行机会 if(orderedChildren.size() == 0){ childWatcher.await(); continue; } for(String headNode : orderedChildren.values()){ String path = dir +"/"+headNode; try{ // 返回对列头部第一个元素 byte[] data = zookeeper.getData(path, false, null); // 删除该元素 zookeeper.delete(path, -1); return data; }catch(KeeperException.NoNodeException e){ // Another client deleted the node first. } } } }

 

private class LatchChildWatcher implements Watcher {        CountDownLatch latch;        public LatchChildWatcher(){            latch = new CountDownLatch(1);        }        public void process(WatchedEvent event){            LOG.debug("Watcher fired on path: " + event.getPath() + " state: " +                     event.getState() + " type " + event.getType());            latch.countDown();        }        public void await() throws InterruptedException {            latch.await();        }    }

 

4. offer  写入队列尾部  使用永久顺序节点

 

public boolean offer(byte[] data) throws KeeperException, InterruptedException{        for(;;){            try{            	// 写入对列    节点类型  永久顺序节点                zookeeper.create(dir+"/"+prefix, data, acl, CreateMode.PERSISTENT_SEQUENTIAL);                return true;            }catch(KeeperException.NoNodeException e){                zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT);            }        }    }

5.peek  返回对列头部  第一个znode对应的值

 

 

public byte[] peek() throws KeeperException, InterruptedException{        try{        	// 返回对列头部第一个znode的值            return element();        }catch(NoSuchElementException e){            return null;        }    }

6. 返回队列头部第一个znode对应的值  并删除该znode

 

 

public byte[] poll() throws KeeperException, InterruptedException {        try{        	// 返回对列znode下 第一个子节点值  并删除该节点            return remove();        }catch(NoSuchElementException e){            return null;        }    }

 

 

 

 

转载地址:http://qpdia.baihongyu.com/

你可能感兴趣的文章
Shell命令-文件及内容处理之sort、uniq
查看>>
Android 之文件夹排序
查看>>
Java Assert 用法简介
查看>>
关于redo size(一)
查看>>
We Know What @You #Tag: Does the Dual Role Affect Hashtag Adoption-20160520
查看>>
(转)Eclipse新增安卓虚拟机
查看>>
SpringMvc访问Controller去掉do
查看>>
PHPnow升级PHP 5.4与Mysql 5.5
查看>>
正则表达式验证邮箱格式
查看>>
如何围绕企业战略,建设BI驾驶舱?
查看>>
java多线程stop,suspend使用代码实际例子
查看>>
中小型研发团队架构实践三:微服务架构(MSA)
查看>>
Windows动态库学习心得
查看>>
在VMware虚拟机上安装Ubuntu 10.04
查看>>
LDA主题模型简介
查看>>
可拖动的DIV续
查看>>
关于“类型初始值设定项引发异常”
查看>>
MySql 小表驱动大表
查看>>
Redis 数据结构的底层实现 (一) RealObject,embstr,sds,ziplist,quicklist
查看>>
SQL语句注入的问题
查看>>