ZMQ学习笔记

亦凉 2022-10-06 11:56 302阅读 0赞

# ZMQ简介 #

官方： “ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层，像框架一样的一个socket library，他使得Socket编程更加简单、简洁和性能更高。ZMQ是一个消息处理队列库，可在多个线程、内核和主机之间弹性伸缩。ZMQ的明确目标是“成为标准网络协议栈的一部分，之后进入Linux内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD套接字之上的一 层封装。ZMQ让编写高性能网络应用程序极为简单和有趣。”

与其他消息中间件相比，ZMQ并不像是一个传统意义上的消息队列服务器，事实上，它也根本不是一个服务器，它更像是一个底层的网络通讯库，在Socket API之上做了一层封装，将网络通讯、进程通讯和线程通讯抽象为统一的API接口。

# ZMQ是什么 #

ZeroMQ命名的确很巧妙。它不是一个消息队列，没有服务器或你需要下载并运行的第三方依赖。事实上，ZeroMQ没有集中的组件。换句话说，ZeroMQ的名字很恰当：ZeroMQ中有零个消息队列零。那它是什么？

ZeroMQ是一个网络库。你创建不同类型的socket（REQ、REP、DEALER、ROUTER、PUB、SUB……），以各种有趣的方式连接这些socket，发送信息到这些socket。socket的类型决定了消息如何路由和传递。

不要以为ZeroMQ是ActiveMQ或RabbitMQ的一种替代方案，或者认为ZeroMQ是一个原子支持网络或socket库。

# 使用套接字创建连接 #

ZMQ提供了一套类似于BSD套接字的API，但将很多消息处理机制的细节隐藏了起来，你会逐渐适应这种变化，并乐于用它进行编程。

套接字事实上是用于网络编程的标准接口，ZMQ之所那么吸引人眼球，原因之一就是它是建立在标准套接字API之上。因此，ZMQ的套接字操作非常容易理解，其生命周期主要包含四个部分：

1.  创建和销毁套接字：zmq\_socket(), zmq\_close()
2.  配置和读取套接字选项：zmq\_setsockopt(), zmq\_getsockopt()
3.  为套接字建立连接：zmq\_bind(), zmq\_connect()
4.  发送和接收消息：zmq\_send(), zmq\_recv()

ZMQ连接和传统的TCP连接是有区别的，主要有：

*  使用多种协议，inproc（进程内）、ipc（进程间）、tcp、pgm（广播）、epgm；
 *  当客户端使用zmq\_connect()时连接就已经建立了，并不要求该端点已有某个服务使用zmq\_bind()进行了绑定；
 *  连接是异步的，并由一组消息队列做缓冲；
 *  连接会表现出某种消息模式，这是由创建连接的套接字类型决定的；
 *  一个套接字可以有多个输入和输出连接；
 *  ZMQ没有提供类似zmq\_accept()的函数，因为当套接字绑定至端点时它就自动开始接受连接了；
 *  应用程序无法直接和这些连接打交道，因为它们是被封装在ZMQ底层的。

ZMQ提供了一组单播传输协议（inporc, ipc, tcp），和两个广播协议（epgm, pgm）

一般而言我们会使用tcp作为传输协议，这种TCP连接是可以脱机运作的，它灵活、便携、且足够快速。为什么称之为脱机，是因为ZMQ中的TCP连接不需要该端点已经有某个服务进行了绑定，客户端和服务端可以随时进行连接和绑定，这对应用程序而言都是透明的。

进程间协议，即ipc，和tcp的行为差不多，但已从网络传输中抽象出来，不需要指定IP地址或者域名,这种协议很多时候会很方便。ZMQ中的ipc协议同样可以是脱机的，但有一个缺点——无法在Windows操作系统上运作，这一点也许会在未来的ZMQ版本中修复。我们一般会在端点名称的末尾附上.ipc的扩展名，在UNIX系统上，使用ipc协议还需要注意权限问题。你还需要保证所有的程序都能够找到这个ipc端点。

进程内协议，即inproc，可以在同一个进程的不同线程之间进行消息传输，它比ipc或tcp要快得多。这种协议有一个要求，必须先绑定到端点，才能建立连接，也许未来也会修复。通常的做法是先启动服务端线程，绑定至端点，后启动客户端线程，连接至端点。

# 使用套接字传递数据 #

发送和接收消息使用的是zmq\_send()和zmq\_recv()这两个函数。虽然函数名称看起来很直白，但由于ZMQ的I/O模式和传统的TCP协议有很大不同，因此还是需要花点时间去理解的。

普通的socket是端到端的（1:1的关系），而ZMQ却是可以N：M 的关系，人们对BSD套接字的了解较多的是点对点的连接，点对点连接需要显式地建立连接、销毁连接、选择协议（TCP/UDP）和处理错误等，而ZMQ屏蔽了这些细节，让你的网络编程更为简单。ZMQ用于node与node间的通信，node可以是主机或者是进程。

![在这里插入图片描述][4fa51b6a05637d760f8399c5495659e9.png_pic_center]  
![在这里插入图片描述][b04961d17dddad6d929772aa4654726f.png_pic_center]

看TCP套接字和ZMQ套接字之间在传输数据方面的区别：

*  ZMQ套接字传输的是消息，而不是字节（TCP）或帧（UDP）。消息指的是一段指定长度的二进制数据块，这种设计是为了性能优化而考虑的，所以可能会比较难以理解。
 *  ZMQ套接字在后台进行I/O操作，也就是说无论是接收还是发送消息，它都会先传送到一个本地的缓冲队列，这个内存队列的大小是可以配置的。
 *  ZMQ套接字可以和多个套接字进行连接（如果套接字类型允许的话）。TCP协议只能进行点对点的连接，而ZMQ则可以进行一对多（类似于无线广播）、多对多（类似于邮局）、多对一（类似于信箱），当然也包括一对一的情况。
 *  ZMQ套接字可以发送消息给多个端点（扇出模型），或从多个端点中接收消息（扇入模型）

## ZMQ消息 ##

ZMQ的传输单位是消息，即一个二进制块。你可以使用任意的序列化工具，如谷歌的Protocal Buffers、XDR、JSON等，将内容转化成ZMQ消息。不过这种转化工具最好是便捷和快速的。

关于消息，还有一些需要注意的地方：

*  ZMQ的消息是作为一个整体来收发的，你不会只收到消息的一部分；
 *  ZMQ不会立即发送消息，而是有一定的延迟；
 *  你可以发送0字节长度的消息，作为一种信号；
 *  消息必须能够在内存中保存，如果你想发送文件或超长的消息，就需要将他们切割成小块，在独立的消息中进行发送；
 *  必须使用zmq\_msg\_close()函数来关闭消息，但在一些会在变量超出作用域时自动释放消息对象的语言中除外。

### 多帧消息 ###

ZMQ消息可以包含多个帧，这在实际应用中非常常见，特别是那些有关“信封”的应用。  
多帧消息的每一帧都是一个zmq\_msg结构，也就是说，当你在收发含有五个帧的消息时，你需要处理五个zmq\_msg结构。你可以将这些帧放入一个数据结构中，或者直接一个个地处理它们。  
下面的代码演示如何发送多帧消息：

zmq_send (socket, &message, ZMQ_SNDMORE);
    ...
    zmq_send (socket, &message, ZMQ_SNDMORE);
    ...
    zmq_send (socket, &message, 0);

然后我们看看如何接收并处理这些消息，这段代码对单帧消息和多帧消息都适用：

while (1) { 
        zmq_msg_t message;
        zmq_msg_init (&message);
        zmq_recv (socket, &message, 0);
        // 处理一帧消息
        zmq_msg_close (&message);
        int64_t more;
        size_t more_size = sizeof (more);
        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
        if (!more)
            break; // 已到达最后一帧
    }

关于多帧消息，你需要了解的还有：

*  在发送多帧消息时，只有当最后一帧提交发送了，整个消息才会被发送；
 *  如果使用了zmq\_poll()函数，当收到了消息的第一帧时，其它帧其实也已经收到了；
 *  多帧消息是整体传输的，不会只收到一部分；
 *  多帧消息的每一帧都是一个zmq\_msg结构；
 *  无论你是否检查套接字的ZMQ\_RCVMORE选项，你都会收到所有的消息；
 *  发送时，ZMQ会将开始的消息帧缓存在内存中，直到收到最后一帧才会发送；
 *  我们无法在发送了一部分消息后取消发送，只能关闭该套接字。

# 核心消息模式 #

## 应答模式 ##

将一组服务端和一组客户端相连，用于远程过程调用或任务分发。  
![在这里插入图片描述][a994b89230aad12788f05f6dd70d5680.png_pic_center]

使用REQ-REP套接字发送和接受消息是需要遵循一定规律的。客户端首先使用zmq\_send()发送消息，再用zmq\_recv()接收，如此循环。如果打乱了这个顺序（如连续发送两次）则会报错。类似地，服务端必须先进行接收，后进行发送。

### 高级应答模式 ###

#### 消息信封 ####

在请求-应答模式中，信封里保存了应答目标的位置。这就是为什么ZMQ网络虽然是无状态的，但仍能完成请求-应答的过程。  
在一般使用过程中，你并不需要知道请求-应答信封的工作原理。使用REQ、REP时，ZMQ会自动处理消息信封。

以下是信封机制在ROUTER中的工作原理：

*  从ROUTER中读取一条消息时，ZMQ会包上一层信封，上面注明了消息的来源。
 *  向ROUTER写入一条消息时（包含信封），ZMQ会将信封拆开，并将消息递送给相应的对象。

如果将从ROUTER A中获取的消息（包含信封）写入ROUTER B（即将消息发送给一个DEALER，该DEALER连接到了ROUTER），那么在从ROUTER B中获取该消息时就会包含两层信封。

信封机制的根本作用是让ROUTER知道如何将消息递送给正确的应答目标，你需要做的就是在程序中保留好该信封。回顾一下REP套接字，它会将收到消息的信封逐个拆开，将消息本身传送给应用程序。而在发送时，又会在消息外层包裹该信封，发送给ROUTER，从而传递给正确的应答目标。

我们可以使用上述原理建立起一个ROUTER-DEALER装置：

[REQ] <--> [REP]
    [REQ] <--> [ROUTER--DEALER] <--> [REP]
    [REQ] <--> [ROUTER--DEALER] <--> [ROUTER--DEALER] <--> [REP]
    ...etc.

当你用REQ套接字去连接ROUTER套接字，并发送一条请求消息，你会从ROUTER中获得一条如下所示的消息：  
![在这里插入图片描述][c7ab9674fb0c01a0e08263b265d502e2.png_pic_center]

*  第三帧是应用程序发送给REQ套接字的消息；
 *  第二帧的空信息是REQ套接字在发送消息给ROUTER之前添加的；
 *  第一帧即信封，是由ROUTER套接字添加的，记录了消息的来源。

如果我们在一条装置链路上传递该消息，最终会得到包含多层信封的消息。最新的信封会在消息的顶部。  
![在这里插入图片描述][d012a4477aafde65495b478255ff6aee.png_pic_center]

以下将详述我们在请求-应答模式中使用到的四种套接字类型：

*  DEALER是一种负载均衡，它会将消息分发给已连接的节点，并使用公平队列的机制处理接受到的消息。DEALER的作用就像是PUSH和PULL的结合。
 *  REQ发送消息时会在消息顶部插入一个空帧，接受时会将空帧移去。其实REQ是建立在DEALER之上的，但REQ只有当消息发送并接受到回应后才能继续运行。
 *  ROUTER在收到消息时会在顶部添加一个信封，标记消息来源。发送时会通过该信封决定哪个节点可以获取到该条消息。
 *  REP在收到消息时会将第一个空帧之前的所有信息保存起来，将原始信息传送给应用程序。在发送消息时，REP会用刚才保存的信息包裹应答消息。REP其实是建立在ROUTER之上的，但和REQ一样，必须完成接受和发送这两个动作后才能继续。

#### 路由 ####

ROUTER是怎么标识消息的来源的？答案当然是套接字的标识。一个套接字可能是瞬时的，它所连接的套接字（如ROUTER）则会给它生成一个标识，与之相关联。一个套接字也可以显式地给自己定义一个标识，这样其他套接字就可以直接使用了。

近距离观察一下REQ和REP套接字，赋予他们一些鲜活的角色：

*  REQ是一个“妈妈”套接字，不会耐心听别人说话，但会不断地抛出问题寻求解答。REQ是严格同步的，它永远位于消息链路的请求端；
 *  REP则是一个“爸爸”套接字，只会回答问题，不会主动和别人对话。REP也是严格同步的，并一直位于应答端。

ROUTER是ZMQ中唯一一个可以定位消息来源的套接字。

##### ROUTER-DEALER路由 #####

将ROUTER和多个DEALER相连接，用一种合适的算法来决定如何分发消息给DEALER。DEALER可以是一个黑洞（只负责处理消息，不给任何返回）、代理（将消息转发给其他节点）或是服务（会发送返回信息）。

如果你要求DEALER能够进行回复，那就要保证只有一个ROUTER连接到DEALER，因为DEALER并不知道哪个特定的节点在联系它，如果有多个节点，它会做负载均衡，将消息分发出去。但如果DEALER是一个黑洞，那就可以连接任何数量的节点。

ROUTER-DEALER路由可以用来做什么呢？如果DEALER会将它完成任务的时间回复给ROUTER，那ROUTER就可以知道这个DEALER的处理速度有多快了。因为ROUTER和DEALER都是异步的套接字，所以我们要用zmq\_poll()来处理这种情况。

+--------+  +--------+  +--------+
    | Client |  | Client |  | Client |
    +--------+  +--------+  +--------+
    |  REQ   |  |  REQ   |  |  REQ   |
    +---+----+  +---+----+  +---+----+
        |           |           |
        +-----------+-----------+
                    |
                +---+----+
                | ROUTER |
                +--------+
                | Device |
                +--------+
                | DEALER |
                +---+----+
                    |
        +-----------+-----------+
        |           |           |
    +---+----+  +---+----+  +---+----+
    |  REP   |  |  REP   |  |  REP   |
    +--------+  +--------+  +--------+
    | Worker |  | Worker |  | Worker |
    +--------+  +--------+  +--------+

这种结构的关键在于，ROUTER会将消息来自哪个REQ记录下来，生成一个信封。DEALER和REP套接字在传输消息的过程中不会丢弃或更改信封的内容，这样当消息返回给ROUTER时，它就知道应该发送给哪个REQ了。这个模型中的REP套接字是匿名的，并没有特定的地址，所以只能提供同一种服务。

上述结构中，对REP的路由我们使用了DEADER自带的负载均衡算法。但是，我们想用LRU算法来进行路由，这就要用到ROUTER-REQ模式：  
![在这里插入图片描述][1cc0d672cd0f8ee351b2a16f192bccd9.png_pic_center]  
LRU算法，它要求client和worker都使用REQ套接字，并正确的存储和返回消息信封，具体如下：

*  创建一组poll，不断地从backend（worker端的ROUTER）获取消息；只有当有空闲的worker时才从frontend（client端的ROUTER）获取消息；
 *  循环执行poll
 *  如果backend有消息，只有两种情况：1）READY消息（该worker已准备好，等待分配）；2）应答消息（需要转发给client）。两种情况下我们都会保存worker的地址，放入LRU队列中，如果有应答内容，则转发给相应的client。
 *  如果frontend有消息，我们从LRU队列中取出下一个worker，将该请求发送给它。这就需要发送worker地址client地址\[请求内容\]到worker端的ROUTER。

#### 实战：跨代理路由 ####

让我们把目前所学到的知识综合起来，应用到实战中去。我们的大客户今天打来一个紧急电话，说是要构建一个大型的云计算设施。它要求这个云架构可以跨越多个数据中心，每个数据中心包含一组client和worker，且能共同协作。

##### 单个集群的架构 #####

worker和client是同步的，我们使用LRU算法来给worker分配任务。每个worker都是等价的，所以我们不需要考虑服务的问题。worker是匿名的，client不会和某个特定的worker进行通信，因而我们不需要保证消息的送达以及失败后的重试等。

鉴于上文提过的原因，client和worker是不会直接通信的，这样一来就无法动态地添加和删除节点了。所以，我们的基础模型会使用一个请求-应答模式中使用过的代理结构。  
![在这里插入图片描述][f76a29120950796bb0f302ff7b3d83b0.png_pic_center]

##### 多个集群的架构 #####

下面我们将集群扩充到多个，每个集群有自己的一组client和worker，并使用代理相连接：

![在这里插入图片描述][11ebcdc6021576e9f595c8739abf92fe.png_pic_center]  
问题在于：我们如何让一个集群的client和另一个集群的worker进行通信呢？

代理之间可以互相连接，这看上去不错，因为不需要增加过多的额外连接。虽然我们不能随意地添加代理，但这个问题可以暂不考虑。这种情况下，集群中的worker和client不必理会整体架构，当代理有剩余的工作能力时便会和其他代理通信。  
![在这里插入图片描述][d0f5a6fbdda7d9cf572c5026cd0c39f2.png_pic_center]  
我们对基础模型应该已经很熟悉了：

*  client线程使用REQ套接字，将请求发送给代理线程（ROUTER套接字）；
 *  worker线程使用REQ套接字，处理并应答从代理线程（ROUTER套接字）收到的请求；
 *  代理会使用LRU队列和路由机制来管理请求。

###### 联邦模式和同伴模式 ######

连接代理的方式有很多，我们需要斟酌一番。我们需要的功能是告诉其他代理“我这里还有空闲的worker”，然后开始接收并处理一些任务；我们还需要能够告诉其他代理“够了够了，我这边的工作量也满了”。

最简单的方式称为联邦，即代理充当其他代理的client和worker。我们可以将代理的前端套接字连接至其他代理的后端套接字，反之亦然。提示一下，ZMQ中是可以将一个套接字绑定到一个端点，同时又连接至另一个端点的。  
![在这里插入图片描述][495e1c023fdbc6dc36d21bcb9582d28c.png_pic_center]

这种机制太简单了，联邦模式下的代理一次只能处理一个请求。如果client和worker是严格同步的，那么代理中的其他空闲worker将分配不到任务。我们需要的代理应该具备完全异步的特性。

但是，联邦模式对某些应用来说是非常好的，比如面向服务架构（SOA）。所以，先不要急着否定联邦模式，它只是不适用于LRU算法和集群负载均衡而已。

我们还有一种方式来连接代理：同伴模式。代理之间知道彼此的存在，并使用一个特殊的信道进行通信。我们逐步进行分析，假设有N个代理需要连接，每个代理则有N-1个同伴，所有代理都使用相同格式的消息进行通信。

关于消息在代理之间的流通有两点需要注意：

*  每个代理需要告知所有同伴自己有多少空闲的worker，这是一则简单的消息，只是一个不断更新的数字，很显然我们会使用PUB-SUB套接字。这样一来，每个代理都会打开一个PUB套接字，不断告知外界自身的信息；同时又会打开一个SUB套接字，获取其他代理的信息。
 *  每个代理需要以某种方式将工作任务交给其他代理，并能获取应答，这个过程需要是异步的。我们会使用ROUTER-ROUTER套接字来实现，没有其他选择。每个代理会使用两个这样的ROUTER套接字，一个用于接收任务，另一个用于分发任务。如果不使用两个套接字，那就需要额外的逻辑来判别收到的是请求还是应答，这就需要在消息中加入更多的信息。

另外还需要考虑的是代理和本地client和worker之间的通信。

代理中有三个消息流，每个消息流使用两个套接字，因此一共需要使用六个套接字。  
以下是我们使用的三个消息流：

*  本地（local）的请求-应答消息流，实现代理和client、代理和worker之间的通信；
 *  云端（cloud）的请求-应答消息流，实现代理和其同伴的通信；
 *  状态（state）流，由代理和其同伴互相传递。

每个消息流会有两个套接字，我们之前一直称为“前端（frontend）”和“后端（backend）”。这两个名字我们已经使用很多次了：前端会负责接受信息或任务；后端会发送信息或任务给同伴。从概念上说，消息流都是从前往后的，应答则是从后往前。

因此，我们决定使用以下的命名方式：

*  localfe / localbe
 *  cloudfe / cloudbe
 *  statefe / statebe

通信协议方面，我们全部使用ipc。

以下是代理程序的套接字分布图：  
![在这里插入图片描述][052f2b7a2a6d1b17ebf2cc765926b2dd.png_pic_center]  
请注意，我们会将cloudbe连接至其他代理的cloudfe，也会将statebe连接至其他代理的statefe。

状态流原型  
![在这里插入图片描述][a05113f9f72c37ecfb9cb183ab7d44ba.png_pic_center]  
几点说明：

*  每个代理都需要有各自的标识，用以生成相应的ipc端点名称。真实环境中，代理需要使用TCP协议连接，这就需要一个更为完备的配置机制，我们会在以后的章节中谈到。
 *  程序的核心是一个zmq\_poll()循环，它会处理接收到消息，并发送自身的状态。只有当zmq\_poll()因无法获得同伴消息而超时时我们才会发送自身状态，如果我们每次收到消息都去发送自身状态，那消息就会过量了。
 *  发送的状态消息包含两帧，第一帧是代理自身的地址，第二帧是空闲的worker数。我们必须要告知同伴代理自身的地址，这样才能接收到请求，唯一的方法就是在消息中显示注明。
 *  我们没有在SUB套接字上设置标识，否则就会在连接到同伴代理时获得过期的状态信息。
 *  我们没有在PUB套接字上设置阈值（HWM），因为订阅者是瞬时的。我们也可以将阈值设置为1，但其实是没有必要的。

本地流和云端流原型  
![在这里插入图片描述][891f4e04c93047bac5862312c91c4b4e.png_pic_center]  
我们需要两个队列，一个队列用于存放从本地集群client收到的请求，另一个存放其他集群发送来的请求。一种方法是从本地和云端的前端套接字中获取消息，分别存入两个队列。但是这么做似乎是没有必要的，因为ZMQ套接字本身就是队列。所以，我们直接使用ZMQ套接字提供的缓存来作为队列使用。

我们的主循环会做以下几件事：

*  轮询后端套接字，会从worker处获得“已就绪”的消息或是一个应答。如果是应答消息，则将其路由回集群client，或是其他集群。
 *  worker应答后即可标记为可用，放入队列并计数；
 *  如果有可用的worker，就获取一个请求，该请求可能来自集群内的client，也可能是其他集群。随后将请求转发给集群内的worker，或是随机转发给其他集群。

## 订阅发布模式 ##

将一组发布者和一组订阅者相连，用于数据分发。  
![在这里插入图片描述][86ade7398a636b465050542a06f66a12.png_pic_center]

第二种经典的消息模式是单向数据分发：服务端将更新事件发送给一组客户端

PUB-SUB套接字组合是异步的。客户端在一个循环体中使用recv ()接收消息，如果向SUB套接字发送消息则会报错；类似地，服务端可以不断地使用send ()发送消息，但不能再PUB套接字上使用recv ()。

关于PUB-SUB套接字，还有一点需要注意：你无法得知SUB是何时开始接收消息的。就算你先打开了SUB套接字，后打开PUB发送消息，这时SUB还是会丢失一些消息的，因为建立连接是需要一些时间的。很少，但并不是零。解决此问题需要在PUB端加入sleep，让PUB延迟一段时间再发送消息。

## 基于分布式处理（管道模式） ##

使用扇入或扇出的形式组装多个节点，可以产生多个步骤或循环，用于构建并行处理架构。  
![在这里插入图片描述][7b4f69b6439b43553a9656c1b0881530.png_pic_center]

1.  任务分发器会生成大量可以并行计算的任务；
2.  有一组worker会处理这些任务；
3.  结果收集器会在末端接收所有worker的处理结果，进行汇总。

## 合法的套接字连接-绑定对（一端绑定、一端连接即可） ##

*  PUB - SUB
 *  REQ - REP
 *  REQ - ROUTER
 *  DEALER - REP
 *  DEALER - ROUTER
 *  DEALER - DEALER
 *  ROUTER - ROUTER
 *  PUSH - PULL
 *  PAIR - PAIR

# ZMQ多线程编程 #

使用ZMQ进行多线程编程时，不需要考虑互斥、锁、或其他并发程序中要考虑的因素，你唯一要关心的仅仅是线程之间的消息。

*  不要在不同的线程之间访问同一份数据，如果要用到传统编程中的互斥机制，那就有违ZMQ的思想了。唯一的例外是ZMQ上下文对象，它是线程安全的。
 *  必须为进程创建ZMQ上下文，并将其传递给所有你需要使用inproc协议进行通信的线程；
 *  你可以将线程作为单独的任务来对待，使用自己的上下文，但是这些线程之间就不能使用inproc协议进行通信了。这样做的好处是可以在日后方便地将程序拆分为不同的进程来运行。
 *  不要在不同的线程之间传递套接字对象，这些对象不是线程安全的。从技术上来说，你是可以这样做的，但是会用到互斥和锁的机制，这会让你的应用程序变得缓慢和脆弱。唯一合理的情形是，在某些语言的ZMQ类库内部，需要使用垃圾回收机制，这时可能会进行套接字对象的传递。
 *  退出程序时，关闭所有的套接字，并在主程序中关闭上下文对象。
 *  如果仍有处于阻塞状态的recv或poll调用，应该在主程序中捕捉这些错误，并在相应的线程中关闭套接字。不要重复关闭上下文，zmq\_term()函数会等待所有的套接字安全地关闭后才结束。

# 中间件和装置 #

当网络组件的数量较少时，所有节点都知道其它节点的存在。但随着节点数量的增加，这种结构的成本也会上升。因此，我们需要将这些组件拆分成更小的模块，使用一个中间件来连接它们。

这种结构在现实世界中是非常常见的，我们的社会和经济体系中充满了中间件的机制，用以降低复杂度，压缩构建大型网络的成本。中间件也会被称为批发商、分包商、管理者等等。

ZMQ网络也是一样，如果规模不断增长，就一定会需要中间件。ZMQ中，我们称其为“装置”。在构建ZMQ软件的初期，我们会画出几个节点，然后将它们连接起来，不使用中间件：  
![在这里插入图片描述][8221b3ca08cbfb98993d73f9d691b8fc.png_pic_center]  
随后，我们对这个结构不断地进行扩充，将装置放到特定的位置，进一步增加节点数量：  
![在这里插入图片描述][0b479d9e13b717e1a03ef32cf75f19e3.png_pic_center]  
ZMQ装置没有具体的设计规则，但一般会有一组“前端”端点和一组“后端”端点。装置是无状态的，因此可以被广泛地部署在网络中。你可以在进程中启动一个线程来运行装置，或者直接在一个进程中运行装置。ZMQ内部也提供了基本的装置实现可供使用。

ZMQ装置可以用作路由和寻址、提供服务、队列调度、以及其他你所能想到的事情。不同的消息模式需要用到不同类型的装置来构建网络。如，请求-应答模式中可以使用队列装置、抽象服务；发布-订阅模式中则可使用流装置、主题装置等。

ZMQ装置比起其他中间件的优势在于，你可以将它放在网络中任何一个地方，完成任何你想要的事情。

## 请求-应答代理（类似分布式http服务器模式） ##

在Hello World客户/服务模型中，一个客户端和一个服务端进行通信。但在真实环境中，我们会需要让多个客户端和多个服务端进行通信。关键问题在于，服务端应该是无状态的，所有的状态都应该包含在一次请求中，或者存放其它介质中，如数据库。

在使用REQ和REP套接字时，其请求-应答的会话是严格同步。客户端发送请求，服务端接收请求并发送应答，由客户端接收。如果客户端或服务端中的一个发生问题（如连续两次发送请求），程序就会报错。

幸运的是，我们有DEALER和ROUTER套接字可以胜任这项工作，进行非阻塞的消息收发。DEALER过去被称为XREQ，ROUTER被称为XREP，但新的代码中应尽量使用DEALER/ROUTER这种名称。

下方的简图描述了一个请求-应答模式，REQ和ROUTER通信，DEALER再和REP通信。ROUTER和DEALER之间我们则需要进行消息转发：  
![在这里插入图片描述][239fcffe6acae9b7f14a85e3fadd00d1.png_pic_center]  
请求-应答代理会将两个套接字分别绑定到前端和后端，供客户端和服务端套接字连接。在使用该装置之前，还需要对客户端和服务端的代码进行调整。

*客户端(client in C)*

//
    // Hello world 客户端
    // 连接REQ套接字至 tcp://localhost:5559 端点
    // 发送Hello给服务端，等待World应答
    //
    #include "zhelpers.h"
     
    int main (void) 
    { 
        void *context = zmq_init (1);
     
        // 用于和服务端通信的套接字
        void *requester = zmq_socket (context, ZMQ_REQ);
        zmq_connect (requester, "tcp://localhost:5559");
     
        int request_nbr;
        for (request_nbr = 0; request_nbr != 10; request_nbr++) { 
            s_send (requester, "Hello");
            char *string = s_recv (requester);
            printf ("收到应答 %d [%s]\n", request_nbr, string);
            free (string);
        }
        zmq_close (requester);
        zmq_term (context);
        return 0;
    }

*服务端(server in C)*

//
    // Hello World 服务端
    // 连接REP套接字至 tcp://*:5560 端点
    // 接收Hello请求，返回World应答
    //
    #include "zhelpers.h"
     
    int main (void) 
    { 
        void *context = zmq_init (1);
     
        // 用于何客户端通信的套接字
        void *responder = zmq_socket (context, ZMQ_REP);
        zmq_connect (responder, "tcp://localhost:5560");
     
        while (1) { 
            // 等待下一个请求
            char *string = s_recv (responder);
            printf ("Received request: [%s]\n", string);
            free (string);
     
            // 做一些“工作”
            sleep (1);
     
            // 返回应答信息
            s_send (responder, "World");
        }
        // 程序不会运行到这里，不过还是做好清理工作
        zmq_close (responder);
        zmq_term (context);
        return 0;
    }

*代理程序(broker in C)*

//
    // 简易请求-应答代理
    //
    #include "zhelpers.h"
     
    int main (void) 
    { 
        // 准备上下文和套接字
        void *context = zmq_init (1);
        void *frontend = zmq_socket (context, ZMQ_ROUTER);
        void *backend  = zmq_socket (context, ZMQ_DEALER);
        zmq_bind (frontend, "tcp://*:5559");
        zmq_bind (backend,  "tcp://*:5560");
     
        // 初始化轮询集合
        zmq_pollitem_t items [] = { 
            {  frontend, 0, ZMQ_POLLIN, 0 },
            {  backend,  0, ZMQ_POLLIN, 0 }
        };
        // 在套接字间转发消息
        while (1) { 
            zmq_msg_t message;
            int64_t more;           // 检测多帧消息
     
            zmq_poll (items, 2, -1);
            if (items [0].revents & ZMQ_POLLIN) { 
                while (1) { 
                    // 处理所有消息帧
                    zmq_msg_init (&message);
                    zmq_recv (frontend, &message, 0);
                    size_t more_size = sizeof (more);
                    zmq_getsockopt (frontend, ZMQ_RCVMORE, &more, &more_size);
                    zmq_send (backend, &message, more? ZMQ_SNDMORE: 0);
                    zmq_msg_close (&message);
                    if (!more)
                        break;      // 最后一帧
                }
            }
            if (items [1].revents & ZMQ_POLLIN) { 
                while (1) { 
                    // 处理所有消息帧
                    zmq_msg_init (&message);
                    zmq_recv (backend, &message, 0);
                    size_t more_size = sizeof (more);
                    zmq_getsockopt (backend, ZMQ_RCVMORE, &more, &more_size);
                    zmq_send (frontend, &message, more? ZMQ_SNDMORE: 0);
                    zmq_msg_close (&message);
                    if (!more)
                        break;      // 最后一帧
                }
            }
        }
        // 程序不会运行到这里，不过还是做好清理工作
        zmq_close (frontend);
        zmq_close (backend);
        zmq_term (context);
        return 0;
    }

使用请求-应答代理可以让你的C/S网络结构更易于扩展：客户端不知道服务端的存在，服务端不知道客户端的存在。网络中唯一稳定的组件是中间的代理装置：

![在这里插入图片描述][f7169d6352b13b6983cd9c641b69ca2c.png_pic_center]

### 内置装置 ###

ZMQ提供了一些内置的装置，不过大多数人需要自己手动编写这些装置。内置装置有：

*  QUEUE，可用作请求-应答代理；
 *  FORWARDER，可用作发布-订阅代理服务；
 *  STREAMER，可用作管道模式代理。  
    可以使用zmq\_device()来启动一个装置，需要传递两个套接字给它：

zmq_device (ZMQ_QUEUE, frontend, backend);

启动了QUEUE队列就如同在网络中加入了一个请求-应答代理，只需为其创建已绑定或连接的套接字即可。

代理程序(Message queue broker in C)

//
    // 简单消息队列代理
    // 功能和请求-应答代理相同，但使用了内置的装置
    //
    #include "zhelpers.h"
     
    int main (void) 
    { 
        void *context = zmq_init (1);
     
        // 客户端套接字
        void *frontend = zmq_socket (context, ZMQ_ROUTER);
        zmq_bind (frontend, "tcp://*:5559");
     
        // 服务端套接字
        void *backend = zmq_socket (context, ZMQ_DEALER);
        zmq_bind (backend, "tcp://*:5560");
     
        // 启动内置装置
        zmq_device (ZMQ_QUEUE, frontend, backend);
     
        // 程序不会运行到这里
        zmq_close (frontend);
        zmq_close (backend);
        zmq_term (context);
        return 0;
    }

内置装置会恰当地处理错误，而我们手工实现的代理并没有加入错误处理机制。所以说，当你能够在程序中使用内置装置的时候就尽量用吧。

### 多线程版 ###

原来的服务是单线程的，如果请求较少，自然没有问题。ZMQ的线程可以在一个核心上高速地运行，执行大量的工作。但是，如果有一万次请求同时发送过来会怎么样？因此，现实环境中，我们会启动多个worker线程，他们会尽可能地接收客户端请求，处理并返回应答。

*服务端(Multithreaded server in C)*

{ 
        void *context = zmq_init (1);
     
        // 用于和client进行通信的套接字
        void *clients = zmq_socket (context, ZMQ_ROUTER);
        zmq_bind (clients, "tcp://*:5555");
     
        // 用于和worker进行通信的套接字
        void *workers = zmq_socket (context, ZMQ_DEALER);
        zmq_bind (workers, "inproc://workers");
     
        // 启动一个worker池
        int thread_nbr;
        for (thread_nbr = 0; thread_nbr < 5; thread_nbr++) { 
            pthread_t worker;
            pthread_create (&worker, NULL, worker_routine, context);
        }
        // 启动队列装置
        zmq_device (ZMQ_QUEUE, clients, workers);
     
        // 程序不会运行到这里，但仍进行清理工作
        zmq_close (clients);
        zmq_close (workers);
        zmq_term (context);
        return 0;
    }

所有的代码应该都已经很熟悉了：

*  服务端启动一组worker线程，每个worker创建一个REP套接字，并处理收到的请求。worker线程就像是一个单线程的服务，唯一的区别是使用了inproc而非tcp协议，以及绑定-连接的方向调换了。
 *  服务端创建ROUTER套接字用以和client通信，因此提供了一个TCP协议的外部接口。
 *  服务端创建DEALER套接字用以和worker通信，使用了内部接口（inproc）。
 *  服务端启动了QUEUE内部装置，连接两个端点上的套接字。QUEUE装置会将收到的请求分发给连接上的worker，并将应答路由给请求方。

我们可以在worker中进行任意的操作，包括与其他节点进行通信。消息的流向是这样的：REQ-ROUTER-queue-DEALER-REP。  
![在这里插入图片描述][16d85b936da2ef6f8c27ff09743eca79.png_pic_center]

# 正确使用上下文 #

ZMQ应用程序的一开始总是会先创建一个上下文，并用它来创建套接字。一个进程中只应该创建一个上下文。从技术的角度来说，上下文是一个容器，包含了该进程下所有的套接字，并为inproc协议提供实现，用以高速连接进程内不同的线程。

# 颠覆传统思维 #

传统网络编程的一个规则是套接字只能和一个节点建立连接。虽然也有广播的协议，但毕竟是第三方的。当我们认定“一个套接字 = 一个连接”的时候，我们会用一些特定的方式来扩展应用程序架构：我们为每一块逻辑创建线程，该线程独立地维护一个套接字。

但在ZMQ的世界里，套接字是智能的、多线程的，能够自动地维护一组完整的连接。你无法看到它们，甚至不能直接操纵这些连接。当你进行消息的收发、轮询等操作时，只能和ZMQ套接字打交道，而不是连接本身。所以说，ZMQ世界里的连接是私有的，不对外部开放，这也是ZMQ易于扩展的原因之一。

由于你的代码只会和某个套接字进行通信，这样就可以处理任意多个连接，使用任意一种网络协议。而ZMQ的消息模式又可以进行更为廉价和便捷的扩展。

这样一来，传统的思维就无法在ZMQ的世界里应用了。在你阅读示例程序代码的时候，也许你脑子里会想方设法地将这些代码和传统的网络编程相关联：当你读到“套接字”的时候，会认为它就表示与另一个节点的连接——这种想法是错误的；当你读到“线程”时，会认为它是与另一个节点的连接——这也是错误的。

ref：https://segmentfault.com/a/1190000039059987  
ref：https://segmentfault.com/a/1190000039060039

[4fa51b6a05637d760f8399c5495659e9.png_pic_center]: /images/20221005/67355804852a4b14bb908b25e602f0bf.png
[b04961d17dddad6d929772aa4654726f.png_pic_center]: /images/20221005/71b77bf19a634b49aa25cf13dc6609c3.png
[a994b89230aad12788f05f6dd70d5680.png_pic_center]: /images/20221005/1416d6dfb05a43c0a97e8be1fd3a89d6.png
[c7ab9674fb0c01a0e08263b265d502e2.png_pic_center]: /images/20221005/0478f4f14a9a424cbb1133be523cc901.png
[d012a4477aafde65495b478255ff6aee.png_pic_center]: /images/20221005/d1ea05a43d7545d898123ab3afe730cb.png
[1cc0d672cd0f8ee351b2a16f192bccd9.png_pic_center]: /images/20221005/6ab48c3c1a104af685e655973c4fc6d5.png
[f76a29120950796bb0f302ff7b3d83b0.png_pic_center]: /images/20221005/e1cfe597a8074a43be742ca4adad2bb4.png
[11ebcdc6021576e9f595c8739abf92fe.png_pic_center]: /images/20221005/9260cd9fd7c546b8b25dbc9b1395676a.png
[d0f5a6fbdda7d9cf572c5026cd0c39f2.png_pic_center]: /images/20221005/0bf6559a3f9c49168f7530615b0b22d9.png
[495e1c023fdbc6dc36d21bcb9582d28c.png_pic_center]: /images/20221005/a61db5495d6e4d27a91990a617a51077.png
[052f2b7a2a6d1b17ebf2cc765926b2dd.png_pic_center]: /images/20221005/8ee2d633fa4b42ada7652ee7e7878c9e.png
[a05113f9f72c37ecfb9cb183ab7d44ba.png_pic_center]: /images/20221005/3c265ce23066430d8830a81ca8febd70.png
[891f4e04c93047bac5862312c91c4b4e.png_pic_center]: /images/20221005/254062953ab2415bbd855bed5bf3de7d.png
[86ade7398a636b465050542a06f66a12.png_pic_center]: /images/20221005/df67a2dd0e064a8f9405c6fde3a4d30a.png
[7b4f69b6439b43553a9656c1b0881530.png_pic_center]: /images/20221005/4659d4743c3447c99af1a07848297e92.png
[8221b3ca08cbfb98993d73f9d691b8fc.png_pic_center]: /images/20221005/eec5ee4df68d4ed5b8ea2a3079b4c0de.png
[0b479d9e13b717e1a03ef32cf75f19e3.png_pic_center]: /images/20221005/4ed132fed93e4eedbd6af7ed50e6e8ff.png
[239fcffe6acae9b7f14a85e3fadd00d1.png_pic_center]: /images/20221005/06a74a3e76874b9bab395b3f2460c35b.png
[f7169d6352b13b6983cd9c641b69ca2c.png_pic_center]: /images/20221005/18f487798d0c4e48ac611aad342989d8.png
[16d85b936da2ef6f8c27ff09743eca79.png_pic_center]: /images/20221005/4c5f5fc53b7f47cb9a9bc69a653c84a5.png