用zookeeper实现简单的发布订阅功能
2017-03-15 08:57
375 查看
发布方Provider的代码
package com.zhuyun.release.subscribe;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
public class Provider {
public static void main(String[] args) throws Exception {
Watcher watcher = new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println("触发了" + event.getPath() + "的" + event.getType() + "事件!");
}
};
ZooKeeper zk = new ZooKeeper("192.168.10.203:2181", 20, watcher);
Stat stat = zk.exists("/message", watcher);
if (stat == null) { //假如节点不存在,则先创建节点
zk.create("/message", "hello".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
//向该节点发送消息
zk.setData("/message", "hello world".getBytes(), -1);
zk.close();
}
}
订阅方Consumer的代码:
package com.zhuyun.release.subscribe;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
public class Consumer implements Watcher{
ZooKeeper zk;
String hostPort;
String znode;
public Consumer(String hostPort,String znode) throws Exception{
this.hostPort = hostPort;
this.znode = znode;
zk = new ZooKeeper(hostPort, 3000, this);
//第一次获取节点消息,同时添加watcher
System.out.println("消息内容:" + new String(zk.getData(znode, true, null)));
}
@Override
public void process(WatchedEvent event) {
if (event.getType() == EventType.NodeDataChanged) {
try {
//当节点消息变化时,触发该操作:获取变化后的消息,同时再添加watcher
System.out.println("你有新的消息:" + new String(zk.getData("/message", true, null)));
} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
public static void main(String[] args) throws Exception {
new Consumer("192.168.10.203:2181","/message");
System.in.read();
}
}
首先启动订阅方监听,再启动发布方发布消息,每次发布一条消息,订阅方都能收到该消息,结果如下:
消息内容:我是谁?
你有新的消息:hello
你有新的消息:你好
你有新的消息:hello world
package com.zhuyun.release.subscribe;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
public class Provider {
public static void main(String[] args) throws Exception {
Watcher watcher = new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println("触发了" + event.getPath() + "的" + event.getType() + "事件!");
}
};
ZooKeeper zk = new ZooKeeper("192.168.10.203:2181", 20, watcher);
Stat stat = zk.exists("/message", watcher);
if (stat == null) { //假如节点不存在,则先创建节点
zk.create("/message", "hello".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
//向该节点发送消息
zk.setData("/message", "hello world".getBytes(), -1);
zk.close();
}
}
订阅方Consumer的代码:
package com.zhuyun.release.subscribe;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
public class Consumer implements Watcher{
ZooKeeper zk;
String hostPort;
String znode;
public Consumer(String hostPort,String znode) throws Exception{
this.hostPort = hostPort;
this.znode = znode;
zk = new ZooKeeper(hostPort, 3000, this);
//第一次获取节点消息,同时添加watcher
System.out.println("消息内容:" + new String(zk.getData(znode, true, null)));
}
@Override
public void process(WatchedEvent event) {
if (event.getType() == EventType.NodeDataChanged) {
try {
//当节点消息变化时,触发该操作:获取变化后的消息,同时再添加watcher
System.out.println("你有新的消息:" + new String(zk.getData("/message", true, null)));
} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
public static void main(String[] args) throws Exception {
new Consumer("192.168.10.203:2181","/message");
System.in.read();
}
}
首先启动订阅方监听,再启动发布方发布消息,每次发布一条消息,订阅方都能收到该消息,结果如下:
消息内容:我是谁?
你有新的消息:hello
你有新的消息:你好
你有新的消息:hello world
相关文章推荐
- ZooKeeper应用场景之消息发布订阅的简单代码实现
- Jedis实现发布订阅功能
- linux下使用hiredis异步API实现sub/pub消息订阅和发布的功能
- 利用redis简单实现消息订阅和发布
- JavaScript简单实现发布订阅模式
- node.js+redis+express 实现发布订阅功能
- 简单的订阅发布机制实现(Golang)
- 利用zookeeper的发布/订阅模式实现配置动态变更
- 利用zookeeper实现发布订阅模式
- Jedis实现Publish/Subscribe功能(发布和订阅)
- python 实现redis订阅发布功能
- 利用reids实现发布订阅的功能
- Thrift 个人实战--RPC服务的发布订阅实现(基于Zookeeper服务)
- Zookeeper实现数据的发布和订阅
- Jquery 自定义事件实现发布/订阅的简单实例
- Jedis实现发布订阅功能
- zookeeper订阅与发布实现
- Thrift 个人实战--RPC服务的发布订阅实现(基于Zookeeper服务)
- reids实现订阅/发布功能
- linux下使用hiredis异步API实现sub/pub消息订阅和发布的功能 标签: hiredishiredis异步APIhiredis事件处理redis消息订阅发布redis c接口 2016-