前面的文章介绍了比较简单的Request/Subscribe模式, 这篇文章介绍更为经典的Publish/Subscribe通信模式用来ZeroMQ的实现,其通信方式如下图:

ZeroMQ之Publish/Subscribe (Java)-LMLPHP

客户端(subscriber)向服务器(publisher)订阅消息,然后服务器可以将消息推送到所有订阅了消息的客户端,这里也可以理解为广播吧。。。。

好了,闲话不多说了,直接上用ZeroMQ实现这种通信模式的代码吧:

(1)服务端(publisher):

  1. package pubsub;
  2. import org.zeromq.ZMQ;
  3. public class Publisher {
  4. public static void main(String args[]) {
  5. ZMQ.Context context = ZMQ.context();  //创创建包含一个I/O线程的context
  6. ZMQ.Socket publisher = context.socket(ZMQ.PUB);   //创建一个publisher类型的socket,他可以向所有订阅的subscriber广播数据
  7. publisher.bind("tcp://*:5555");  //将当前publisher绑定到端口上,可以接受subscriber的订阅
  8. while (!Thread.currentThread ().isInterrupted ()) {
  9. String message = "fjs hello";  //最开始可以理解为pub的channel,subscribe需要订阅fjs这个channel才能接收到消息
  10. publisher.send(message.getBytes());
  11. }
  12. publisher.close();
  13. context.term();
  14. }
  15. }

代码很简单吧,这里publisher来充当服务端,所有的subscriber都需要建立于publisher的连接。,,。,

(2)客户端(subscriber)代码:

  1. package pubsub;
  2. import org.zeromq.ZMQ;
  3. public class Subscriber {
  4. public static void main(String args[]) {
  5. for (int j = ; j < ; j++) {
  6. new Thread(new Runnable(){
  7. public void run() {
  8. // TODO Auto-generated method stub
  9. ZMQ.Context context = ZMQ.context();  //创建1个I/O线程的上下文
  10. ZMQ.Socket subscriber = context.socket(ZMQ.SUB);     //创建一个sub类型,也就是subscriber类型的socket
  11. subscriber.connect("tcp://127.0.0.1:5555");    //与在端口监听的publisher建立连接
  12. subscriber.subscribe("fjs".getBytes());     //订阅fjs这个channel
  13. for (int i = ; i < ; i++) {
  14. byte[] message = subscriber.recv();  //接收publisher发送过来的消息
  15. System.out.println("receive : " + new String(message));
  16. }
  17. subscriber.close();
  18. context.term();
  19. }
  20. }).start();
  21. }
  22. }
  23. }

这里需要注意订阅的channel问题,如果这里错了的话,subscriber是不会受到publisher发送过来的数据的

好了,到这里publish/subscribe的实现就算ok了

04-24 03:40