简述

ZeroMQ（ØMQ）也叫ZMQ，是一个github开源项目，他为可伸缩的分布式或并发应用程序设计的高性能异步消息库。它提供一个消息队列，但是与面向消息的中间件不同，ZeroMQ的运行不需要专门的消息代理（message broker）。该库设计成常见的套接字（Socket）风格的API。

  • 优点
    • 高性能
    • 开源
    • 简洁
    • 跨语言
    • 提供多种消息传输协议：TCP，PGM（可靠的多播），进程间通信（IPC） 以及线程间通讯（ITC）
  • 缺点
    • 不具备消息持久化

image.png
image.png

消息模型

请求响应模式（Request-Reply）

将一组客户端连接到一组服务器。这是一种远程过程调用和任务分发模式。

  1. public class ReqZeroMQ {
  2.     public static void main(String[] args) {
  3.         ZMQ.Context context = ZMQ.context(1);
  4.         // Socket to talk to server
  5.         System.out.println("Connecting to hello world server…");
  6.         ZMQ.Socket requester = context.socket(SocketType.REQ);
  7.         requester.connect("tcp://localhost:5555");
  8.         for (int requestNbr = 0; requestNbr != 100; requestNbr++) {
  9.             String request = "Hello";
  10.             System.out.println("Sending Hello " + requestNbr);
  11.             requester.send(request.getBytes(), 0);
  12.             byte[] reply = requester.recv(0);
  13.             System.out.println("Received " + new String(reply) + " " + requestNbr);
  14.         }
  15.         requester.close();
  16.         context.term();
  17.     }
  18. }
  1. public class RepZeroMQ {
  2.     public static void main(String[] args) {
  3.         new Thread(() -> {
  4.             ZMQ.Context context = ZMQ.context(1);
  5.             // Socket to talk to clients
  6.             ZMQ.Socket responder = context.socket(SocketType.REP);
  7.             responder.bind("tcp://*:5555");
  8.             while (!Thread.currentThread().isInterrupted()) {
  9.                 // Wait for next request from the client
  10.                 byte[] request = responder.recv(0);
  11.                 System.out.println("Received Hello");
  12.                 // Do some 'work'
  13.                 try {
  14.                     Thread.sleep(1000);
  15.                 } catch (InterruptedException e) {
  16.                     e.printStackTrace();
  17.                 }
  18.                 // Send reply back to client
  19.                 String reply = "World";
  20.                 responder.send(reply.getBytes(), 0);
  21.             }
  22.             responder.close();
  23.             context.term();
  24.         }).start();
  25.     }
  26. }

发布/订阅模式（Publish-Subscribe）

将一组发布者连接到一组订阅者。这是一种数据分发模式。

  1. public class PubZeroMQ {
  2.     public static void main(String[] args) {
  3.         // Prepare our context and publisher
  4.         ZMQ.Context context = ZMQ.context(1);
  5.         ZMQ.Socket publisher = context.socket(SocketType.PUB);
  6.         publisher.bind("tcp://*:5556");
  8.         // Initialize random number generator
  9.         Random srandom = new Random(System.currentTimeMillis());
  10.         while (!Thread.currentThread().isInterrupted()) {
  11.             // Get values that will fool the boss
  12.             int zipcode, temperature, relhumidity;
  13.             zipcode = 10000 + srandom.nextInt(10000);
  14.             temperature = srandom.nextInt(215) - 80 + 1;
  15.             relhumidity = srandom.nextInt(50) + 10 + 1;
  17.             // Send message to all subscribers
  18.             String update = String.format("%05d %d %d", zipcode, temperature, relhumidity);
  19.             publisher.send(update, 0);
  20.         }
  21.         publisher.close();
  22.         context.term();
  23.     }
  24. }
  1. public class SubZeroMQ {
  2.     public static void main(String[] args) {
  3.         new Thread(() -> {
  4.             ZMQ.Context context = ZMQ.context(1);
  5.             // Socket to talk to server
  6.             System.out.println("Collecting updates from weather server");
  7.             ZMQ.Socket subscriber = context.socket(SocketType.SUB);
  8.             subscriber.connect("tcp://localhost:5556");
  10.             // Subscribe to zipcode, default is NYC, 10001
  11.             String filter = (args.length > 0) ? args[0] : "10001 ";
  12.             subscriber.subscribe(filter.getBytes());
  14.             // Process 100 updates
  15.             int update_nbr;
  16.             long total_temp = 0;
  17.             for (update_nbr = 0; update_nbr < 100; update_nbr++) {
  18.                 // Use trim to remove the tailing '0' character
  19.                 String string = subscriber.recvStr(0).trim();
  20.                 StringTokenizer sscanf = new StringTokenizer(string, " ");
  21.                 int temperature = Integer.valueOf(sscanf.nextToken());
  23.                 total_temp += temperature;
  24.             }
  25.             System.out.println("Average temperature for zipcode '"
  26.                     + filter + "' was " + (int) (total_temp / update_nbr));
  27.             System.out.println("...");
  28.             try {
  29.                 System.in.read();
  30.             } catch (IOException e) {
  31.                 e.printStackTrace();
  32.             }
  33.             subscriber.close();
  34.             context.term();
  35.         }).start();
  36.         new Thread(() -> {
  37.             ZMQ.Context context = ZMQ.context(1);
  38.             // Socket to talk to server
  39.             System.out.println("Collecting updates from weather server");
  40.             ZMQ.Socket subscriber = context.socket(ZMQ.SUB);
  41.             subscriber.connect("tcp://localhost:5556");
  43.             // Subscribe to zipcode, default is NYC, 10001
  44.             String filter = (args.length > 0) ? args[0] : "10002";
  45.             subscriber.subscribe(filter.getBytes());
  47.             // Process 100 updates
  48.             int update_nbr;
  49.             long total_temp = 0;
  50.             for (update_nbr = 0; update_nbr < 100; update_nbr++) {
  51.                 // Use trim to remove the tailing '0' character
  52.                 String string = subscriber.recvStr(0).trim();
  54.                 StringTokenizer sscanf = new StringTokenizer(string, " ");
  55.                 int temperature = Integer.valueOf(sscanf.nextToken());
  57.                 total_temp += temperature;
  58.             }
  59.             System.out.println("Average temperature for zipcode '"
  60.                     + filter + "' was " + (int) (total_temp / update_nbr));
  61.             System.out.println("...");
  62.             try {
  63.                 System.in.read();
  64.             } catch (IOException e) {
  65.                 e.printStackTrace();
  66.             }
  67.             subscriber.close();
  68.             context.term();
  69.         }).start();
  70.     }
  72. }

管道模式（Push-Pull）

以Push/Pull模式连接节点，可以有多个步骤，可以有循环。这是一种并行的任务分发和收集模式。

  1. public class PushZeroMQ {
  2.     public static void main(String[] args) {
  3.         new Thread(() -> {
  4.             ZMQ.Context context = ZMQ.context(1);
  5.             ZMQ.Socket socket = context.socket(SocketType.PUSH);
  6.             socket.connect("tcp://127.0.0.1:5555");
  7.             socket.setSendTimeOut(1000);
  8.             int i = 0;
  9.             while (true) {
  10.                 byte[] data = String.valueOf(i++).getBytes();
  11.                 boolean send = socket.send(data);
  12.                 System.out.println(socket.getTCPKeepAlive() + " send data " + data.length + " result " + send);
  13.             }
  14.         }).start();
  15.     }
  16. }
  1. public class PullZeroMQ {
  2.     public static void main(String[] args) {
  3.         new Thread(() -> {
  4.             ZMQ.Context context = ZMQ.context(1);
  5.             ZMQ.Socket socket = context.socket(SocketType.PULL);
  6.             socket.bind("tcp://127.0.0.1:5555");
  7.             socket.setReceiveTimeOut(1000);
  8.             while (true) {
  9.                 String recv = socket.recvStr();
  10.                 System.out.println(socket.getTCPKeepAlive() + " recv data " + "result " + recv);
  11.             }
  12.         }).start();
  13.     }
  14. }

排他对模式

在一个排他对中连接两个套接字（Socket）。（这是一种高级的为某种用例而设计的低级别模式）

参考文献

消息队列库——ZeroMQ
RabbitMq、ActiveMq、ZeroMq、kafka之间的比较,资料汇总
ØMQ维基百科