為了在Zookeeper中實現分布式隊列，首先需要設計一個znode來存放數據，這個節點叫做隊列節點，我們的例子中這個節點是/zookeeper/queue。 生產者向隊列中存放數據，每一個消息都是隊列節點下的一個新節點，叫做消息節點。消息節點的命名規則為：queue-xxx，xxx是一個單調 遞增的序列，我們可以在創建節點時指定創建模式為PERSISTENT_SEQUENTIAL來實現。這樣，生產者不斷的向隊列節點中發送消息，消息為queue-xxx， 隊列中，生產者這一端就解決了，我們具體看一下代碼：
Producer（生產者）
public class Producer implements Runnable,Watcher {
private ZooKeeper zk;
public Producer(String address){
try {
this.zk = new ZooKeeper(address,3000,this);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
int i = 0;
//每隔10s向隊列中放入數據
while (true){
try {
zk.create("/zookeeper/queue/queue-",(Thread.currentThread().getName()+"-"+i).getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT_SEQUENTIAL);
Thread.sleep(10000);
i++;
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
@Override
public void process(WatchedEvent event) {
}
}
生產者每隔10s向隊列中存放消息，消息節點的類型為PERSISTENT_SEQUENTIAL，消息節點中的數據為Thread.currentThread().getName()+”-“+i。
消費者
消費者從隊列節點中獲取消息，我們使用getChildren()方法獲取到隊列節點中的所有消息，然后獲取消息節點數據，消費消息，並刪除消息節點。 如果getChildren()沒有獲取到數據，說明隊列是空的，則消費者等待，然后再調用getChildren()方法設置觀察者監聽隊列節點，隊列節點發生變化后 （子節點改變），觸發監聽事件，喚起消費者。消費者實現如下：
public class Consumer implements Runnable,Watcher {
private ZooKeeper zk;
private List<String> children;
public Consumer(String address){
try {
this.zk = new ZooKeeper(address,3000,this);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
int i = 1;
while (true){
try {
//獲取所有子節點
children = zk.getChildren("/zookeeper/queue", false);
int size = CollectionUtils.isEmpty(children) ? 0 : children.size();
System.out.println("第"+i+"次獲取數據"+size+"條");
//隊列中沒有數據，設置觀察器並等待
if (CollectionUtils.isEmpty(children)){
System.out.println("隊列為空，消費者等待");
zk.getChildren("/zookeeper/queue", true);
synchronized (this){
wait();
}
}else {
//循環獲取隊列中消息，進行業務處理，並從結果集合中刪除
Iterator<String> iterator = children.iterator();
while (iterator.hasNext()){
String childNode = iterator.next();
handleBusiness(childNode);
iterator.remove();
}
}
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
i++;
}
}
/**
* 從節點獲取數據，執行業務，並刪除節點
* @param childNode
*/
private void handleBusiness(String childNode) {
try {
Stat stat = new Stat();
byte[] data = zk.getData("/zookeeper/queue/"+childNode, false, stat);
String str = new String(data);
System.out.println("獲取節點數據："+str);
zk.delete("/zookeeper/queue/"+childNode,-1);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 子節點發生變化，且取得結果為空時，說明消費者等待，喚起消費者
* @param event
*/
@Override
public void process(WatchedEvent event) {
if (event.getType().equals(Event.EventType.NodeChildrenChanged)){
synchronized (this){
notify();
}
}
}
}
上面的例子中有一個局限性，就是 消費者只能有一個 。隊列的用戶有兩個：廣播和隊列。
- 廣播是所有消費者都拿到消息並消費，我們的例子在刪除消息節點時，不能保證其他消費者都拿到了這個消息。
- 隊列是一個消息只能被一個消費者消費，我們的例子中，消費者獲取消息時，並沒有加鎖。
所以我們只啟動一個消費者來演示，主函數如下：
public class Application {
private static final String ADDRESS = "149.28.37.147:2181";
public static void main(String[] args) {
//設置日志級別
setLog();
//啟動一個消費者
new Thread(new Consumer(ADDRESS)).start();
//啟動4個生產者
ExecutorService es = Executors.newFixedThreadPool(4);
for (int i=0;i<4;i++){
es.execute(new Producer(ADDRESS));
}
es.shutdown();
}
/**
* 設置log級別為Error
*/
public static void setLog(){
//1.logback
LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
//獲取應用中的所有logger實例
List<Logger> loggerList = loggerContext.getLoggerList();
//遍歷更改每個logger實例的級別,可以通過http請求傳遞參數進行動態配置
for (ch.qos.logback.classic.Logger logger:loggerList){
logger.setLevel(Level.toLevel("ERROR"));
}
}
}
后台打印結果如下：
第1次獲取數據2條 獲取節點數據：pool-1-thread-4-118 獲取節點數據：pool-1-thread-1-0 第2次獲取數據3條 獲取節點數據：pool-1-thread-4-0 獲取節點數據：pool-1-thread-2-0 獲取節點數據：pool-1-thread-3-0 第3次獲取數據0條 隊列為空，消費者等待 第4次獲取數據4條 獲取節點數據：pool-1-thread-3-1 獲取節點數據：pool-1-thread-1-1 獲取節點數據：pool-1-thread-4-1 獲取節點數據：pool-1-thread-2-1
Zookeeper實現隊列就介紹完了，項目地址：https://github.com/liubo-tech/zookeeper-application。