zookeeper應用

發布訂閱

zk實現的方式是推拉結合，Client想服務端注冊自己需要關注的節點，一旦節點的數據發生變更，那么Server會向對應的客戶端發送Watcher事件通知，客戶端接收到這個消息后，需要主動到服務端獲取最新的數據。

目前很多應用使用發布訂閱都不是用zk的這種方式，比較典型的純的推模式和拉模式，這個之前有記錄過Notify和MetaQ的比較，不是本篇的重點。本次主要是利用zookeeper來實現以下發布訂閱這種功能。

搭建了一個zk環境，手動創建了一個節點/publish,客戶端發布者代碼如下：

package com.wpr.zk.pulish;import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat;import java.io.IOException; import java.util.concurrent.CountDownLatch;/*** 利用zk來模擬發布訂閱模式* Created by peirong.wpr on 2017/4/5.*/ public class Publish implements Watcher{private static CountDownLatch latch = new CountDownLatch(1);private static Stat stat = new Stat();private static ZooKeeper zk =null;private final static Integer SESSION_TIMEOUT = 5000;public static void main(String[] args) {try {String path ="/publish";zk = new ZooKeeper("192.168.109.130:2181",SESSION_TIMEOUT,new Publish());latch.await();System.out.println("zk connection");byte[] temp = zk.getData(path,true,stat);System.out.println("init data :pulish node data"+new String(temp));int i=0;while(true){System.out.println( "publish new Data:"+i);zk.setData(path,String.valueOf(i).getBytes(),-1);Thread.sleep(5000L);i++;}} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();}}public void process(WatchedEvent event) {if(Event.KeeperState.SyncConnected == event.getState()){System.out.println("receive watched event:"+event);System.out.println(event.getState());latch.countDown();}} }

訂閱者代碼如下：

package com.wpr.zk.pulish;import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat;import java.io.IOException; import java.util.concurrent.CountDownLatch;/*** Created by peirong.wpr on 2017/4/5.*/ public class Subscribe implements Watcher {private static CountDownLatch latch = new CountDownLatch(1);private static Stat stat = new Stat();private static ZooKeeper zk =null;private final static Integer SESSION_TIMEOUT = 5000;public static void main(String[] args) {try {String path ="/publish";zk = new ZooKeeper("192.168.109.130:2181",SESSION_TIMEOUT,new Subscribe());latch.await();System.out.println("zk connection");byte[] temp = zk.getData(path,true,stat);System.out.println("init data :pulish node data"+new String(temp));int i=0;while(true){Thread.sleep(Integer.MAX_VALUE);}} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();}}public void process(WatchedEvent event) {if(Event.KeeperState.SyncConnected == event.getState()){if(Event.EventType.None == event.getType() && event.getPath() == null){latch.countDown();}else if(event.getType() == Event.EventType.NodeDataChanged){//Clinet需要去拉取最新的數據信息try {byte[] newByte = zk.getData(event.getPath(),true,stat);System.out.println("path:"+event.getPath()+"\tdata has changed.\t new Data :"+ new String(newByte));} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}}} }

轉載于:https://www.cnblogs.com/kakaxisir/p/6667279.html

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

以上是生活随笔為你收集整理的利用zookeeper实现发布订阅模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。