ZeroMQ模式介绍

ZeroMQ的模式分类

ZeroMQ模式在zmq_socket()接口中有介绍，可以参阅：http://api.zeromq.org/master:zmq-socket

zmq_socket

zmq_socket - 创建0MQ套接字

void *zmq_socket (void *context, int type);

zmq_socket()函数将在指定的上下文中创建一个ZeroMQ套接字，并返回新创建的套接字

type 类型参数指定套接字类型，它确定套接字上通信的语义。

MQ套接字是异步的，这意味着物理连接设置和拆卸、重新连接和有效传递的时间对用户是透明的，并且是由MQ本身组织的。此外，如果对等方无法接收消息，则可能会排队。

与传统套接字的关键区别

一般来说，传统的套接字是 1、面向连接的可靠字节流(SOCK_STREAM) 或者 2、不可靠的数据报(SOCK_DGRAM)，并为其提供同步接口。

相比之下，ZeroMQ套接字是对异步消息队列的抽象。根据使用的套接字type 类型的不同，而具有不同的模式。传统套接字传输字节流或离散数据报，ZeroMQ套接字传输离散消息。

ZeroMQ套接字是异步的，意味着物理连接的建立与断开，重连，数据有效传输，对用户是透明的，并ZeroMQ组织处理。此外，如果对方无法接收消息，那么消息可能会排队。

传统的套接字只允许严格的一对一(两个对等点)、多对一(多个客户机、一个服务器)，或者在某些情况下只允许的一对多(多播) 关系。除了ZMQ_PAIR之外，ZeroMQ套接字可以使用ZMQ_CONNECT() http://api.zeromq.org/master:zmq-connect连接到多个端点，同时接受来自使用ZMQ_BIND() http://api.zeromq.org/master:zmq-bind 绑定到套接字的多个端点的传入连接，从而允许多到多关系。

Socket types 类型

type 设置不同的值决定了ZeroMQ Socket 的不同类型，也就是ZeroMQ的不同模式：

客户端-服务器模式（Client-server）、广播盘模式（Radio-dish）、发布订阅模式（Publish-subscribe）、流水线模式（Pipeline pattern）、独占对模式（Exclusive pair）、本机模式（Native）、请求-回复模式（Request-reply）

客户端-服务器模式（Client-server）

types 类型：ZMQ_CLIENT 、ZMQ_SERVER
客户机-服务器模式用于允许一个ZMQ_SERVER服务器与一个或多个ZMQ_CLIENT客户机通信。客户端总是启动对话，之后任何一方都可以向另一方异步发送消息
客户机-服务器模式由http://rfc.zeromq.org/spec:41正式定义，该模式仍在草案阶段

广播盘模式（Radio-dish）

types 类型：ZMQ_RADIO、ZMQ_DISH
广播盘模式用于从单个发行者到多个订阅者的一对多的数据分发。
广播碟仍处于草案阶段

发布订阅模式（Publish-subscribe）

types 类型：ZMQ_PUB、ZMQ_SUB、ZMQ_XPUB、ZMQ_XSUB
发布-订阅模式用于从单个发布服务器到多个订阅者的一对多的数据分发。
发布-订阅模式由http://rfc.zeromq.org/spec:29正式定义

ZMQ_PUB

发布者使用类型为ZMQ_PUB的套接字来分发数据。发送的消息以扇形的方式分发给所有连接的订阅者。
对于此套接字类型，不能执行zmq_recv() http://api.zeromq.org/master:zmq-recv函数。

ZMQ_SUB

订阅者使用ZMQ_SUB类型的套接字来订阅发布者分发的数据
ZMQ_SUB套接字创建完成之后，ZMQ_SUB套接字未订阅任何消息，请使用zmq_setsockopt() http://api.zeromq.org/master:zmq-setsockopt的ZMQ_SUBSCRIBE选项指定要订阅的消息

ZMQ_XPUB

用法与ZMQ_PUB大部分相同，但是有一点与ZMQ_PUB不同。ZMQ_XPUB(自己)的订阅方可以向自己发送一个订阅信息来进行订阅。订阅消息是字节1（用于订阅）或字节0（用于取消订阅），后跟订阅主体。

ZMQ_XSUB

用法与ZMQ_SUB大部分相同，但是有一点与ZMQ_SUB不同。自己可以向发布者发送一条订阅信息来进行订阅。订阅消息是字节1（用于订阅）或字节0（用于取消订阅），后跟订阅主体。

示例代码:

使用：zeromq/libzmq 的 libzmq 4.3.3

API接口： http://api.zeromq.org

GitHub： PublishSubscribeDemo

发布者代码：

//  Binds PUB socket to tcp://*:5556
//  Publishes random weather updates
#include "zhelpers.h"int main (void)
{//  Prepare our context and publishervoid *context = zmq_ctx_new ();void *publisher = zmq_socket (context, ZMQ_PUB);int rc = zmq_bind (publisher, "tcp://*:5556");assert (rc == 0);//  Initialize random number generatorsrandom ((unsigned) time (NULL));while (1) {//  Get values that will fool the bossint zipcode, temperature, relhumidity;zipcode     = randof (100000);temperature = randof (215) - 80;relhumidity = randof (50) + 10;//  Send message to all subscriberschar update [20];sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);s_send (publisher, update);}zmq_close (publisher);zmq_ctx_destroy (context);return 0;
}

订阅者代码：

//  Weather update client
//  Connects SUB socket to tcp://localhost:5556
//  Collects weather updates and finds avg temp in zipcode
#include "zhelpers.h"int main (int argc, char *argv [])
{//  Socket to talk to serverprintf ("Collecting updates from weather server...\n");void *context = zmq_ctx_new ();void *subscriber = zmq_socket (context, ZMQ_SUB);int rc = zmq_connect (subscriber, "tcp://localhost:5556");assert (rc == 0);//  Subscribe to zipcode, default is NYC, 10001const char *filter = (argc > 1)? argv [1]: "10001 ";rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,filter, strlen (filter));assert (rc == 0);//  Process 100 updatesint update_nbr;long total_temp = 0;for (update_nbr = 0; update_nbr < 100; update_nbr++) {char *string = s_recv (subscriber);int zipcode, temperature, relhumidity;sscanf (string, "%d %d %d",&zipcode, &temperature, &relhumidity);total_temp += temperature;free (string);}printf ("Average temperature for zipcode '%s' was %dF\n",filter, (int) (total_temp / update_nbr));zmq_close (subscriber);zmq_ctx_destroy (context);return 0;
}

流水线模式（Pipeline pattern）

types 类型：ZMQ_PUSH、ZMQ_PULL
所述流水线模式用于将数据分配给排列在管道中的节点。数据总是沿着管道流动，管道的每个阶段都连接到至少一个节点。当一个管道连接到多个节点时，数据在所有连接节点之间被循环。
流水线模式由http://rfc.zeromq.org/spec:30正式定义

ZMQ_PUSH

管道节点使用ZMQPush类型的套接字向下游管道节点发送消息。消息被循环到所有连接的下游节点。不能执行zmq_recv() http://api.zeromq.org/master:zmq-recv。

ZMQ_PULL

管道节点使用ZMQ_PULL类型的套接字从上游管道节点接收消息。消息从所有连接的上游节点中公平排队。不能执行zmq_send() http://api.zeromq.org/master:zmq-send。

独占对模式（Exclusive pair）

types 类型：ZMQ_PAIR
独占对模式用于将对等点连接到另一个对等点。
独占对模式由http://rfc.zeromq.org/spec:31正式定义

ZMQ_PAIR

ZMQ_PAIR类型的套接字在任何时候只能连接到一个对等点。不对通过ZMQ_PAIR套接字发送的消息执行任何消息路由或筛选。

本机模式（Native）

types 类型：ZMQ_STREAM
本机模式用于与TCP对等点通信，并允许任何方向的异步请求和答复。

请求-回复模式（Request-reply）

types 类型：ZMQ_REQ、ZMQ_REP、ZMQ_DEALER、ZMQ_ROUTER
Request-Reply模式用于从ZMQ_REQ客户端向一个或多个ZMQ_REP服务发送请求，并接收发送的每个请求的后续答复。

ZMQ_REQ

客户端使用ZMQ_REQ类型的套接字向服务发送请求并接收服务的答复。此套接字类型只允许交替的ZMQ_Send和ZMQ_recv调用。如果没有可用的服务，则套接字上的任何发送操作都应阻塞，直到至少有一项服务可用为止。REQ套接字不会丢弃消息。

ZMQ_REP

服务使用ZMQ_REP类型的套接字来接收来自客户端的请求并发送答复。此套接字类型只允许交替的ZMQ_Send和ZMQ_recv调用。如果原始请求者不再存在，则答复将被静默丢弃。

ZMQ_DEALER

ZMQ_ARIRER类型的套接字是用于扩展请求/应答套接字的高级模式。

ZMQ_ROUTER

ZMQ_ROUTER类型的套接字是用于扩展请求/应答套接字的高级模式。

示例代码:

使用：zeromq/libzmq 的 libzmq 4.3.3

API接口： http://api.zeromq.org

GitHub： HelloWorld

server代码：

//  Hello World server#include <zmq.h>
#include <stdio.h>
#include <unistd.h>
#include <assert.h>int main (void)
{int major, minor, patch;zmq_version(&major, &minor, &patch); // 获取当前使用的ZeroMQ的版本号printf("Current ZeroMQ version is %d.%d.%d\n", major, minor, patch);///  1.创建上下文void *context = zmq_ctx_new ();   // 创建 0MQ上下文  线程安全的/// // 2.创建、绑定套接字// ZMQ_REP 服务使用ZMQ_REP类型的套接字来接收来自客户端的请求并向客户端发送回复void *responder = zmq_socket (context, ZMQ_REP);int rc = zmq_bind (responder, "tcp://*:5555");assert (rc == 0);/// 3.循环接收数据、发送数据while (1) {char buffer [10];/// 4.接收数据zmq_recv (responder, buffer, 10, 0);printf ("Received Hello\n");sleep (1);/// 5.回送数据zmq_send (responder, "World", 5, 0);}return 0;
}

client代码：

//  Hello World client
#include <zmq.h>
#include <stdio.h>int main (void)
{printf ("Connecting to hello world server...\n");/// 1、创建上下文void *context = zmq_ctx_new ();   // 创建 0MQ上下文  线程安全的/// 2.创建、绑定套接字// ZMQ_REQ  客户端使用ZMQ_REQ类型的套接字向服务发送请求并从服务接收答复void *requester = zmq_socket (context, ZMQ_REQ);zmq_connect (requester, "tcp://localhost:5555");///3.循环发送数据、接收数据int request_nbr;for (request_nbr = 0; request_nbr != 10; request_nbr++) {char buffer [10];printf ("Sending Hello %d...\n", request_nbr);/// 4.发送数据zmq_send (requester, "Hello", 5, 0);// 5.接收回复数据zmq_recv (requester, buffer, 10, 0);printf ("Received World %d\n", request_nbr);}/// 6.关闭套接字、销毁上下文zmq_close (requester);zmq_ctx_destroy (context);return 0;
}