散仙,在前几篇关于zookeeper的文章中,介绍了基于zookeeper实现的分布式公平锁,以及非公平锁,那么本篇呢,散仙就来介绍下关于使用zookeeper如何模拟实现一个分布式队列。
那么为什么需要分布式队列呢?,我们都知道队列,在我们的编程开发中,是一种比不可少的数据结构,最典型莫过于,生产者与消费者的例子了,我们在程序过经常使用的队列是基于非分布式的环境,JAVA JDK也自带了非常多的队列的实现,有基于阻塞模式的,也有基于非阻塞模式的,这些我们可以在不同的场景下使用。
要想把JDK自带的队列给设计成分布式的队列,是一件非常繁琐的事,这时候我就可以轻而易举的使用zookeeper来完成这个功能,zookeeper由于其特殊的文件系统,所以在分布式环境下,能做许多有用的事。
例如在Hadoop中,map任务完成后,我们启动reduce,我们都知道Hadoop的MapReduce是一种分布式的计算框架,所以这时候我们就可以使用zookeeper来完成这里的分布式队列的运用,虽然,hadoop本身用的并不是zookeeper来实现的。
代码如下:
package com.qin.queue;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import com.test.Lock1;
/**
*
* 基于zookeeper实现的
* 分布式队列
* @author qindongliang
*
* **/
public class DistributeQueue1 {
private ZooKeeper zk;
private CountDownLatch down=new CountDownLatch(1);
public DistributeQueue1(String zkHost) {
try{
zk=new ZooKeeper(zkHost, 5000, new Watcher(){
@Override
public void process(WatchedEvent event) {
if(event.getState()==Event.KeeperState.SyncConnected){
down.countDown();
}
}
});
}catch(Exception e){
e.printStackTrace();
}
}
/**
* 创建一个持久node,
*
* **/
public void createPersist()throws Exception{
zk.create("/a", "主节点".getBytes(), Ids.OPEN_ACL_UNSAFE , CreateMode.PERSISTENT);
System.out.println("创建主节点成功........");
}
/**
*
* 校验队列情况
*
* **/
public void check()throws Exception{
List<String> list=zk.getChildren("/a", null);
if(list.size()==3){
execute();
}else{
System.out.println("阻塞中......,队列成员数量:"+list.size());
}
}
public void execute(){
System.out.println("队列已满,开始执行任务......");
}
/***
*
* 注册队列
*
*
* */
public void createTemp()throws Exception{
zk.create("/a/b", "a".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("有任务注册队列成功");
check();
}
public static void main(String[] args)throws Exception {
DistributeQueue1 dq=new DistributeQueue1("192.168.120.128:2181");
//dq.createPersist();
dq.createTemp();
Thread.sleep(Integer.MAX_VALUE);
}
}
代码如上,散仙在代码里设定队列的总数为3,就执行干某一件事,我可以去zk的主节点上注册临时有序的节点,每注册一次都会校验,当前队列是不是以及满队,如果满队的话,就开始执行某个任务,从而达到分布式队列的模拟,这与JDK自带的CountDownLatch和CyclicBarrier是非常类似的,当然我们也可也使用zk模拟,这两个非分布式的功能,从而为我们的应用程序在分布式的环境下提供便利。
分享到:
相关推荐
分布式之Zookeeper(高清视频教程).rar
ubuntu系统部署完全分布式hadoop时,先部署了zookeeper和hbase
分布式框架Zookeeper之服务注册与订阅.........................
分布式架构Zookeeper入门实战到底层原理剖析
Zookeeper伪分布式集群搭建描述伪分布式集群就是在一台机器部署多个zk应用,部署之前服务器需要有jdk环境 java -version可以显示相关java
互联网大厂必会的高薪技能之分布式架构Zookeeper入门实战到底层原理剖析
server程序入口(启动类在zookeeper-server/src/main/java文件夹中,org.apache.zookeeper.server.quorum.QuorumPeerMain),program argument为conf/zoo.cfg,将conf/log4j.properties配置拷贝到zookeeper-server/...
三台主机搭建分布式Zookeeper,完全分布式搭建zookeeper集群,及其简单的操作
分布式服务框架 Zookeeper -- 管理分布式环境中的数据
本文来自于cnblogs,说到分布式开发Zookeeper是必须了解和掌握的,分布式消息服务kafka、hbase到hadoop等分布式大数据处理都会用到Zookeeper,所以在此将Zookeeper作为基础来讲解。Zookeeper是分布式服务框架,主要...
#资源达人分享计划#
详解Zookeeper伪分布式集群环境搭建过程
基于zookeeper的分布式锁简单实现,包含测试代码,实用工具类
ZooKeeper 是一个开源的分布式 应用程序协调服务器,其为分布式系统提供一致性服务。其一致性是通过基于 Paxos 算法的 ZAB 协议完成的。其主要功能包括:配置维护、域名服务、分布式同步、集群管理等。 zookeeper 的...
#资源达人分享计划#
从Paxos到Zookeeper分布式一致性原理与实践 + ZooKeeper-分布式过程协同技术详解
在分布式高并发的情况下,多个访问之间拥有唯一的一个订单编号,多次提交的订单不会出现重复的,只能是唯一的,ZooKeeper就很好的解决了唯一性
从传统JavaWeb转入分布式系统应用,再到接触分布式协调框架ZooKeeper,通过痛苦的思维逻辑和理念转变,历经一个月时间,小伙伴们终于把ZooKeeper嵌入到了BoCloud博云的BeyondContainer中,并在其上进行相应功能的...
分布式服务框架zookeeper 3.4.13版本,主要用于构建RPC服务与常见集群等。