ZeroMQ之Publish/Subscribe (Java)
2016-04-28 09:44
531 查看
前面的文章介绍了比较简单的Request/Subscribe模式, 这篇文章介绍更为经典的Publish/Subscribe通信模式用来ZeroMQ的实现,其通信方式如下图:
客户端(subscriber)向服务器(publisher)订阅消息,然后服务器可以将消息推送到所有订阅了消息的客户端,这里也可以理解为广播吧。。。。
好了,闲话不多说了,直接上用ZeroMQ实现这种通信模式的代码吧:
(1)服务端(publisher):
[java] view plaincopy
package pubsub;
import org.zeromq.ZMQ;
public class Publisher {
public static void main(String args[]) {
ZMQ.Context context = ZMQ.context(1); //创创建包含一个I/O线程的context
ZMQ.Socket publisher = context.socket(ZMQ.PUB); //创建一个publisher类型的socket,他可以向所有订阅的subscriber广播数据
publisher.bind("tcp://*:5555"); //将当前publisher绑定到5555端口上,可以接受subscriber的订阅
while (!Thread.currentThread ().isInterrupted ()) {
String message = "fjs hello"; //最开始可以理解为pub的channel,subscribe需要订阅fjs这个channel才能接收到消息
publisher.send(message.getBytes());
}
publisher.close();
context.term();
}
}
代码很简单吧,这里publisher来充当服务端,所有的subscriber都需要建立于publisher的连接。,,。,
(2)客户端(subscriber)代码:
[java] view plaincopy
package pubsub;
import org.zeromq.ZMQ;
public class Subscriber {
public static void main(String args[]) {
for (int j = 0; j < 100; j++) {
new Thread(new Runnable(){
public void run() {
// TODO Auto-generated method stub
ZMQ.Context context = ZMQ.context(1); //创建1个I/O线程的上下文
ZMQ.Socket subscriber = context.socket(ZMQ.SUB); //创建一个sub类型,也就是subscriber类型的socket
subscriber.connect("tcp://127.0.0.1:5555"); //与在5555端口监听的publisher建立连接
subscriber.subscribe("fjs".getBytes()); //订阅fjs这个channel
for (int i = 0; i < 100; i++) {
byte[] message = subscriber.recv(); //接收publisher发送过来的消息
System.out.println("receive : " + new String(message));
}
subscriber.close();
context.term();
}
}).start();
}
}
}
这里需要注意订阅的channel问题,如果这里错了的话,subscriber是不会受到publisher发送过来的数据的
好了,到这里publish/subscribe的实现就算ok了
客户端(subscriber)向服务器(publisher)订阅消息,然后服务器可以将消息推送到所有订阅了消息的客户端,这里也可以理解为广播吧。。。。
好了,闲话不多说了,直接上用ZeroMQ实现这种通信模式的代码吧:
(1)服务端(publisher):
[java] view plaincopy
package pubsub;
import org.zeromq.ZMQ;
public class Publisher {
public static void main(String args[]) {
ZMQ.Context context = ZMQ.context(1); //创创建包含一个I/O线程的context
ZMQ.Socket publisher = context.socket(ZMQ.PUB); //创建一个publisher类型的socket,他可以向所有订阅的subscriber广播数据
publisher.bind("tcp://*:5555"); //将当前publisher绑定到5555端口上,可以接受subscriber的订阅
while (!Thread.currentThread ().isInterrupted ()) {
String message = "fjs hello"; //最开始可以理解为pub的channel,subscribe需要订阅fjs这个channel才能接收到消息
publisher.send(message.getBytes());
}
publisher.close();
context.term();
}
}
代码很简单吧,这里publisher来充当服务端,所有的subscriber都需要建立于publisher的连接。,,。,
(2)客户端(subscriber)代码:
[java] view plaincopy
package pubsub;
import org.zeromq.ZMQ;
public class Subscriber {
public static void main(String args[]) {
for (int j = 0; j < 100; j++) {
new Thread(new Runnable(){
public void run() {
// TODO Auto-generated method stub
ZMQ.Context context = ZMQ.context(1); //创建1个I/O线程的上下文
ZMQ.Socket subscriber = context.socket(ZMQ.SUB); //创建一个sub类型,也就是subscriber类型的socket
subscriber.connect("tcp://127.0.0.1:5555"); //与在5555端口监听的publisher建立连接
subscriber.subscribe("fjs".getBytes()); //订阅fjs这个channel
for (int i = 0; i < 100; i++) {
byte[] message = subscriber.recv(); //接收publisher发送过来的消息
System.out.println("receive : " + new String(message));
}
subscriber.close();
context.term();
}
}).start();
}
}
}
这里需要注意订阅的channel问题,如果这里错了的话,subscriber是不会受到publisher发送过来的数据的
好了,到这里publish/subscribe的实现就算ok了
相关文章推荐
- android stadio使用eclipse的debug.keystore
- Intellij idea 开发第一个springMVC demo
- Java中集合Set的用法
- Spring MVC Web框架
- Java里面String.equalsIgnoreCase()方法和java.lang.String.equals()方法的区别
- 前端工程师的Java路--Spring MVC 初使用
- Struts迭代器(iterator)遍历List常用的4种例子
- 在eclipse中如何设置字体的大小
- 详解Java设计模式编程中命令模式的项目结构实现
- JAVA基础知识
- eclipse中一些常见svn图标的含义
- Java第八次实验要求
- IntelliJ IDEA 创建spring mvc项目(图)
- java中堆和栈的区别
- spring-源码学习-1
- Java.lang.UnsatisfiedLinkError
- hadoop2.7.2 win7 eclipse环境搭建测试
- struts2与hibernate的优缺点
- JAVA日志系统
- eclipse经常弹出Subversion Native Library Not Available,什么是JavaHL,svnkit完美解决