ZMQv3.2指南（Python版）第二章

在第一章 - 基础中，我们体验了 ZeroMQ ，并学习了一些主要 ZeroMQ 模式（请求-回复、发布-订阅和管道）的基本示例。在本章中，我们将动手深入学习如何在真实程序中使用这些工具。

我们将覆盖：

让我们来看一下TCP套接字和ZeroMQ套接字在处理数据方面的主要差异：

ZeroMQ不是一个中立的载体：它对其使用的传输协议施加了一个框架。这个框架与现有协议不兼容，现有协议往往使用自己的框架。例如，在TCP / IP上比较HTTP请求和ZeroMQ请求。HTTP请求使用CR-LF作为其最简单的框架定界符，而ZeroMQ使用长度指定的帧。因此，您可以使用ZeroMQ编写类似HTTP的协议，例如使用请求-回复套接字模式。但它不是HTTP。

ZeroMQ有一个名为ZMQROUTERRAW的套接字选项，它允许你在没有ZeroMQ封装的情况下读写数据。您可以使用它读写正确的HTTP请求和响应。Hardeep Singh提出了这个更改，以便他可以从他的ZeroMQ应用程序连接到Telnet服务器。截至编写本文时，它仍然是有点实验性的，但它显示了ZeroMQ如何不断发展以解决新的问题。I/O 线程 ZeroMQ在后台线程中执行I/O。除了极端的应用程序外，一个I/O线程(对于所有套接字)就足够了。当您创建一个新的上下文时，它将以一个I/O线程开始。通用经验法则是每秒入或出的数据量为一千兆字节允许一个I/O线程。要增加I/O线程的数量，请在创建任何套接字之前使用zmqctxset()调用：int io_threads = 4;void *context = zmq_ctx_new ();zmq_ctx_set (context, ZMQ_IO_THREADS, io_threads);assert (zmq_ctx_get (context, ZMQ_IO_THREADS) == io_threads); 我们已经看到，一个套接字可以同时处理数十甚至数千个连接。这对于如何编写应用程序有着根本的影响。传统的网络应用程序每个远程连接有一个进程或一个线程，该进程或线程处理一个套接字。 ZeroMQ允许您将整个结构压缩到单个进程中，然后根据需要进行分割以实现扩展。 如果您仅使用ZeroMQ进行线程间通信（即，不进行外部套接字I / O的多线程应用程序），则可以将I / O线程设置为零。这不是一个重要的优化，而是一个好奇。消息模式 ZeroMQ在套接字API的棕色纸片下面隐藏着消息模式的世界。如果您有企业消息的背景知识，或者熟悉UDP，这些将是熟悉的。但对于大多数ZeroMQ的新手，它们是一个惊喜。我们习惯于将套接字一一映射到另一个节点的TCP范式。 让我们简要回顾一下ZeroMQ为您做了什么。它快速高效地将数据块（消息）传递给节点。您可以将节点映射到线程、进程或节点。无论实际传输（例如进程内、进程间、TCP或多播）如何，ZeroMQ都给您的应用程序提供单一的套接字API。当对等方出现和消失时，它会自动重新连接到对等方。在发送方和接收方需要时，它会对队列中的消息进行队列。它将这些队列限制为保护进程免受内存不足的影响。它处理套接字错误。它在后台线程中执行所有I/O。它使用无锁技术与节点通信，因此永远不会有锁定、等待、信号量或死锁。 然而，通过它，它根据称为模式的精确配方对消息进行路由和排队。正是这些模式提供了ZeroMQ的智能。它们封装了我们对分发数据和工作的最佳方法的经验。 ZeroMQ的模式是硬编码的，但未来的版本可能允许用户定义的模式。 ZeroMQ模式由具有匹配类型的套接字对实现。换句话说，要理解ZeroMQ模式，您需要理解套接字类型以及它们如何一起工作。大部分只需学习；在这个层面上几乎没有什么是显而易见的。 核心的 ZeroMQ 模式包括：

不要现在使用 zmqmsginit_data()。这是一种零复制方法，保证会给你带来麻烦。在你开始担心减少微秒前，有更多重要的关于 ZeroMQ 的东西需要学习。

这个丰富的 API 可能很难使用。这些方法是为了性能而优化，而不是简单易用。如果开始使用它们，在读完手册页之前，你几乎肯定会把它们做错。因此，一个好的语言绑定的主要任务是将这个 API 包装在更易于使用的类中。

到目前为止，所有示例的主循环都是：

如果我们想同时从多个端点读取，最简单的方法是将一个套接字连接到所有端点，让 ZeroMQ 为我们完成扇入。如果远程端点在同一模式中，这是合法的，但将 PULL 套接字连接到 PUB 端点是错误的。

要同时从多个套接字读取，请使用 poller.poll() 。另一种更好的方法可能是将 poller.poll() 包装在一个框架中，将其变成一个很好的事件驱动的反应器，但这比我们想要覆盖的内容要多得多。

让我们从一个脏技巧开始，一部分是为了不做正确的乐趣，但主要是因为它可以让我向您展示如何进行非阻塞套接字读取。下面是一个使用非阻塞读取从两个套接字读取的简单示例。这个相当困惑的程序既作为对天气更新的订阅者，又作为并行任务的工作者：

msreader.py

# encoding: utf-8## Reading from multiple sockets# This version uses a simple recv loop## Author: Jeremy Avnet (brainsik) #

import zmqimport time

# Prepare our context and socketscontext = zmq.Context()

# Connect to task ventilatorreceiver = context.socket(zmq.PULL)receiver.connect("tcp://localhost:5557")

# Connect to weather serversubscriber = context.socket(zmq.SUB)subscriber.connect("tcp://localhost:5556")subscriber.setsockopt(zmq.SUBSCRIBE, b"10001")

# Process messages from both sockets# We prioritize traffic from the task ventilatorwhile True:

# Process any waiting tasks while True: try: msg = receiver.recv(zmq.DONTWAIT) except zmq.Again: break # process task

# Process any waiting weather updates while True: try: msg = subscriber.recv(zmq.DONTWAIT) except zmq.Again: break # process weather update

# No activity, so sleep for 1 msec time.sleep(0.001)

这种方法的代价是在第一条消息上的额外延迟（循环末尾的睡眠，当没有等待处理的消息时）。这将是在子毫秒级延迟至关重要的应用中的问题。此外，您还需要检查您使用的任何函数的文档，以确保它不会造成繁忙循环。

您可以通过先从一个读取，再从第二个读取，而不是将它们作为我们在此示例中所做的那样优先处理，来公平对待套接字。

现在让我们看一下使用 poller.poll() 正确完成的相同的无意义小应用程序：

mspoller.py

# encoding: utf-8## Reading from multiple sockets# This version uses zmq.Poller()## Author: Jeremy Avnet (brainsik) #

import zmq

# Prepare our context and socketscontext = zmq.Context()

# Connect to task ventilatorreceiver = context.socket(zmq.PULL)receiver.connect("tcp://localhost:5557")

# Connect to weather serversubscriber = context.socket(zmq.SUB)subscriber.connect("tcp://localhost:5556")subscriber.setsockopt(zmq.SUBSCRIBE, b"10001")

# Initialize poll setpoller = zmq.Poller()poller.register(receiver, zmq.POLLIN)poller.register(subscriber, zmq.POLLIN)

# Process messages from both socketswhile True: try: socks = dict(poller.poll()) except KeyboardInterrupt: break

if receiver in socks: message = receiver.recv() # process task

if subscriber in socks: message = subscriber.recv() # process weather update

结构体 items 具有以下四个成员：

typedef struct { void *socket; // 要轮询的ZeroMQ套接字 int fd; // 或者要轮询的原生文件句柄 short events; // 要轮询的事件 short revents; // 轮询后返回的事件} zmq_pollitem_t;

ZeroMQ 允许我们将一条消息分成多帧构成“多帧消息”。现实应用中经常大量使用多帧消息，既用于将消息与地址信息一起封装，也用于简单的序列化。我们稍后会讲回复信封。

我们现在将学习的是在任意应用程序（如代理）中盲目且安全地读写多帧消息，而无需检查它们。

当你使用多帧消息时，每个部分都是一个 zmqmsg 项。例如，如果要发送五部分的消息，您必须构造、发送和销毁五个 zmqmsg 项。您可以提前进行（并将 zmq_msg 项存储在数组或其他结构中），或者逐个发送。

以下是我们发送多帧消息中的帧的方法（我们将每个帧接收到消息对象）：

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

这是如何接收和处理消息的所有部分的方法，无论是单个部分还是多个部分：

while (1) { zmq_msg_t message; zmq_msg_init (&message); zmq_msg_recv (&message, socket, 0); // Process the message frame ... zmq_msg_close (&message); if (!zmq_msg_more (&message)) break; // Last message frame}

关于多段消息，你需要了解的一些事情：

实际上，这会导致架构变得越来越不稳定和难以管理。假设您有一个发布者和一百个订阅者。您通过在每个订阅者中配置发布者端点来将每个订阅者连接到发布者。很容易。订阅者是动态的，发布者是静态的。现在再加入更多的发布者。突然之间，它不再那么容易了。如果继续将每个订阅者与每个发布者连接，避免动态发现的成本就会越来越高。

解决这个问题的方法有很多，但是最简单的方法是加入一个中间件，也就是一个网络中所有其他节点连接的静态点。在传统的消息处理中，这是消息代理的工作。ZeroMQ并没有像消息代理一样的功能，但它很容易构建中间件。 你可能想知道，如果所有网络最终都需要中间件，为什么不直接为所有应用程序提供消息代理呢？对于初学者来说，这是一个合理的妥协。只需始终使用星形拓扑，忽略性能，通常情况下就能正常工作。然而，消息代理是一种贪婪的东西；作为中央中间件，它变得太复杂，太状态性，最终成为问题。 最好将中间件视为简单的无状态消息交换。一个很好的类比是HTTP代理；它在那里，但没有任何特殊角色。添加一个pub-sub代理可以解决我们示例中的动态发现问题。我们将代理设置在网络的“中间”。代理打开一个XSUB套接字和一个XPUB套接字，并将它们各自绑定到已知的IP地址和端口。然后，所有其他进程连接到代理，而不是相互连接。添加更多的订阅者或发布者变得很容易。

我们需要XPUB和XSUB套接字，因为ZeroMQ对订阅者进行订阅转发到发布者。XSUB和XPUB与SUB和PUB完全相同，只是它们将订阅作为特殊消息公开。代理必须从XPUB套接字读取这些订阅消息并将它们写入XSUB套接字，以从订阅者端转发到发布者端。这是XSUB和XPUB的主要用例。共享队列（DEALER 和 ROUTER 套接字） 在 Hello World 客户端/服务端应用程序中，我们有一个客户端与一个服务器通信。但是在实际情况下，我们通常需要允许多个服务和多个客户端。这使我们能够扩展服务的能力（多个线程或进程或节点，而不仅仅是一个）。唯一的限制是服务必须是无状态的，所有状态都在请求中或在某些共享存储中，例如数据库中。

有两种方法将多个客户端连接到多个服务器。暴力方法是将每个客户端套接字连接到多个服务端点。一个客户端套接字可以连接到多个服务套接字，然后 REQ 套接字将请求分配到这些服务中。假设你将一个客户端套接字连接到三个服务端点；A，B 和 C。客户端发送请求 R1，R2，R3，R4。R1 和 R4 去 A 服务，R2 去 B，R3 去 C 服务。 这种设计可以便宜地添加更多客户端。您也可以添加更多服务。每个客户端将向服务分发请求。但是，每个客户端都必须知道服务拓扑。如果您有100个客户端，然后再决定添加三个更多的服务，您需要重新配置和重启100个客户端，以便客户端了解这三个新服务。 这显然不是我们想要在凌晨3点做的事情，当我们的超级计算集群的资源耗尽时，我们急需增加几百个新的服务节点。太多静态的东西就像液态水泥：知识是分散的，你有越多静态的东西，改变拓扑的难度就越大。我们想要的是站在客户端和服务之间的东西，它集中了所有关于拓扑的知识。理想情况下，我们应该能够随时添加和删除服务或客户端，而不影响拓扑的其他部分。 因此，我们将编写一个小的消息队列代理，为我们提供这种灵活性。代理绑定到两个端点，一个是客户端的前端，一个是服务的后端。然后它使用 poller.poll() 监视这两个套接字的活动，并在有活动时在两个套接字之间穿梭消息。它实际上并不明确地管理任何队列，ZeroMQ在每个套接字上自动执行这一操作。 当使用REQ与REP通信时，您将获得严格同步的请求-回复对话。客户端发送请求。服务读取请求并发送回复。客户端然后读取回复。如果客户端或服务试图做任何其他操作（例如，在不等待响应的情况下连续发送两个请求），它们将出错。 但我们的代理必须是非阻塞的。显然，我们可以使用 poller.poll() 等待套接字上的活动，但我们不能使用REP和REQ。

幸运的是，有两个叫做DEALER和ROUTER的套接字，它们允许您执行非阻塞请求-响应。您将在第3章-高级请求-回复模式中看到，DEALER和ROUTER套接字如何帮助您构建各种异步请求-回复流。现在，我们只需看一下DEALER和ROUTER如何扩展REQ-REP跨越中间件，也就是我们的小经纪人。 在这种简单的扩展请求-回复模式中，REQ与ROUTER进行通信，DEALER与REP进行通信。在DEALER和ROUTER之间，我们必须有代码（例如我们的经纪人），从一个套接字上拉消息并将其推到另一个套接字上。 请求-回复经纪人绑定到两个端点，一个是供客户端连接的（前端套接字），一个是供工作者连接的（后端套接字）。为了测试此经纪人，您需要更改工作者，使其连接到后端套接字。这里是一个客户端，显示了我的意思：rrclient.py## Request-reply client in Python# Connects REQ socket to tcp://localhost:5559# Sends "Hello" to server, expects "World" back#import zmq

# Prepare our context and socketscontext = zmq.Context()socket = context.socket(zmq.REQ)socket.connect("tcp://localhost:5559")

# Do 10 requests, waiting each time for a responsefor request in range(1, 11): socket.send(b"Hello") message = socket.recv() print(f"Received reply {request} [{message}]") 这是工作者：rrworker.py## Request-reply service in Python# Connects REP socket to tcp://localhost:5560# Expects "Hello" from client, replies with "World"#import zmq

context = zmq.Context()socket = context.socket(zmq.REP)socket.connect("tcp://localhost:5560")

while True: message = socket.recv() print(f"Received request: {message}") socket.send(b"World") 这是代理，它正确处理多帧消息：rrbroker.py# Simple request-reply broker## Author: Lev Givon

import zmq

# Prepare our context and socketscontext = zmq.Context()frontend = context.socket(zmq.ROUTER)backend = context.socket(zmq.DEALER)frontend.bind("tcp://*:5559")backend.bind("tcp://*:5560")

# Initialize poll setpoller = zmq.Poller()poller.register(frontend, zmq.POLLIN)poller.register(backend, zmq.POLLIN)

# Switch messages between socketswhile True: socks = dict(poller.poll())

if socks.get(frontend) == zmq.POLLIN: message = frontend.recv_multipart() backend.send_multipart(message)

if socks.get(backend) == zmq.POLLIN: message = backend.recv_multipart() frontend.send_multipart(message)

使用请求-回复代理可以使您的客户端/服务器架构更易于扩展，因为客户端看不到工作者，工作者看不到客户端。中间唯一的静态节点是代理。ZeroMQ内置代理功能 事实证明，前面章节中 rrbroker 的核心循环非常有用，并且可重复使用。它可以让我们非常轻松地构建 pub-sub 转发器和共享队列以及其他小型中间件。ZeroMQ 将此包装在单个方法 zmq_proxy() 中：zmq_proxy (frontend, backend, capture); 这两个（如果我们想捕获数据，可能是三个套接字）必须正确连接，绑定和配置。当我们调用zmqproxy方法时，它就像启动rrbroker的主循环一样。让我们重写请求回复代理以调用zmqproxy，并将其重新标记为昂贵的“消息队列”（人们已经因所做的代码收取了费用，它的功能比它更少）：msgqueue.py"""

Simple message queuing broker Same as request-reply broker but using ``zmq.proxy``

Author: Guillaume Aubert (gaubert)

"""

import zmq

def main(): """ main method """

context = zmq.Context()

# Socket facing clients frontend = context.socket(zmq.ROUTER) frontend.bind("tcp://*:5559")

# Socket facing services backend = context.socket(zmq.DEALER) backend.bind("tcp://*:5560")

zmq.proxy(frontend, backend)

# We never get here... frontend.close() backend.close() context.term()

if __name__ == "__main__": main() 如果你像大多数ZeroMQ用户一样，此时你的思想开始想到：“如果我随机插入代理的套接字类型，我可以做什么邪恶的事情？”简短的回答是：试试并弄清楚发生了什么。实际上，通常你会选择ROUTER / DEALER，XSUB / XPUB或PULL / PUSH。运输桥接 ZeroMQ用户经常询问：“如何将我的ZeroMQ网络与技术X连接？”其中X是其他网络或消息传递技术。

简单的答案是构建桥接。桥接是一个小应用程序，在一个套接字上使用一种协议，在另一个套接字上进行第二种协议的转换/转换。一种协议解释器，如果您愿意。 ZeroMQ中的一个常见的桥接问题是桥接两个传输或网络。 例如，我们将编写一个小的代理，它坐在发布者和一组订阅者之间，桥接两个网络。前端套接字（SUB）面向内部网络，其中天气服务器坐在，后端（PUB）面向外部网络的订阅者。它在前端套接字上订阅天气服务，并在后端套接字上重新发布其数据。wuproxy.py# Weather proxy device## Author: Lev Givon

import zmq

context = zmq.Context()

# This is where the weather server sitsfrontend = context.socket(zmq.SUB)frontend.connect("tcp://192.168.55.210:5556")

# This is our public endpoint for subscribersbackend = context.socket(zmq.PUB)backend.bind("tcp://10.1.1.0:8100")

# Subscribe on everythingfrontend.setsockopt(zmq.SUBSCRIBE, b'')

# Shunt messages out to our own subscriberswhile True: # Process all parts of the message message = frontend.recv_multipart() backend.send_multipart(message) 这看起来非常类似于先前的代理示例，但关键部分是前端和后端套接字位于两个不同的网络中。我们可以使用这种模型，例如将多播网络（pgm传输）连接到tcp发布者。处理错误和 ETERM ZeroMQ 的错误处理理念是快速失败和弹性的混合。我们认为，进程对内部错误应该尽可能脆弱，对外部攻击和错误应该尽可能强大。为了给出一个类比，一个活细胞如果检测到单个内部错误，它会自我摧毁，但它将通过一切可能的方法抵抗外部攻击。 断言，它们涂抹着 ZeroMQ 代码，对于稳健的代码至关重要；它们只需要在细胞壁的正确侧面。应该有这样的壁。如果不清楚故障是内部还是外部，这是一个需要修复的设计缺陷。在 C/C++ 中，断言立即以错误的方式终止应用程序。在其他语言中，您可能会得到异常或中止。 当ZeroMQ检测到外部故障时，它会向调用代码返回错误。在某些罕见的情况下，如果没有明显的恢复策略，它可能会静默丢弃消息。 到目前为止，我们看到的大多数C示例都没有错误处理。实际代码应该对每一个ZeroMQ调用都进行错误处理。如果您使用的不是C语言绑定，那么绑定可能会为您处理错误。在C语言中需要自行处理。以POSIX惯例为基础的一些简单规则：

void *context = zmq_ctx_new ();assert (context);void *socket = zmq_socket (context, ZMQ_REP);assert (socket);int rc = zmq_bind (socket, "tcp://*:5555");if (rc == -1) { printf ("E: bind failed: %s\n", strerror (errno)); return -1;}

有两种主要的异常情况，您应该将其作为非致命性处理：

让我们看看如何干净地关闭进程。我们将采用前一节中的并行管道示例。如果我们已经在后台启动了很多工人，现在我们希望在批处理完成时杀死它们。我们可以通过向工人发送杀死消息来实现这一点。这样做的最佳地点是Sink，因为它真的知道批处理何时完成。 我们如何将Sink连接到工人？PUSH / PULL套接字只是单向的。我们可以切换到另一种套接字类型，也可以混合多个套接字流。我们尝试后者：使用 pub-sub 模型向工人发送 kill 消息：

import sysimport timeimport zmq

context = zmq.Context()

# Socket to receive messages onreceiver = context.socket(zmq.PULL)receiver.connect("tcp://localhost:5557")

# Socket to send messages tosender = context.socket(zmq.PUSH)sender.connect("tcp://localhost:5558")

# Socket for control inputcontroller = context.socket(zmq.SUB)controller.connect("tcp://localhost:5559")controller.setsockopt(zmq.SUBSCRIBE, b"")

# Process messages from receiver and controllerpoller = zmq.Poller()poller.register(receiver, zmq.POLLIN)poller.register(controller, zmq.POLLIN)# Process messages from both socketswhile True: socks = dict(poller.poll())

if socks.get(receiver) == zmq.POLLIN: message = receiver.recv_string()

# Process task workload = int(message) # Workload in msecs

# Do the work time.sleep(workload / 1000.0)

# Send results to sink sender.send_string(message)

# Simple progress indicator for the viewer sys.stdout.write(".") sys.stdout.flush()

# Any waiting controller command acts as 'KILL' if socks.get(controller) == zmq.POLLIN: break

# Finishedreceiver.close()sender.close()controller.close()context.term() 这是修改后的接收端应用程序。 当它完成收集结果时，它将 kill 消息广播到所有工作者：# encoding: utf-8## Task sink - design 2# Adds pub-sub flow to send kill signal to workers## Author: Jeremy Avnet (brainsik) #

import sysimport timeimport zmq

context = zmq.Context()

# Socket to receive messages onreceiver = context.socket(zmq.PULL)receiver.bind("tcp://*:5558")

# Socket for worker controlcontroller = context.socket(zmq.PUB)controller.bind("tcp://*:5559")

# Wait for start of batchreceiver.recv()

# Start our clock nowtstart = time.time()

# Process 100 confirmiationsfor task_nbr in range(100): receiver.recv() if task_nbr % 10 == 0: sys.stdout.write(":") else: sys.stdout.write(".") sys.stdout.flush()

# Calculate and report duration of batchtend = time.time()tdiff = tend - tstarttotal_msec = tdiff * 1000print("Total elapsed time: %d msec" % total_msec)

# Send kill signal to workerscontroller.send(b"KILL")

# Finishedreceiver.close()controller.close()context.term()处理中断信号 实际的应用程序需要在用 Ctrl-C 或其他信号（如 SIGTERM）中断时干净地关闭。默认情况下，这些信号只是简单地杀死进程，这意味着消息不会被刷新，文件不会干净地关闭等。 以下是各种语言中处理信号的方法：interrupt.py## Shows how to handle Ctrl-C#import signalimport timeimport zmq

context = zmq.Context()socket = context.socket(zmq.REP)socket.bind("tcp://*:5558")

# SIGINT will normally raise a KeyboardInterrupt, just like any other Python calltry: socket.recv()except KeyboardInterrupt: print("W: interrupt received, stopping...")finally: # clean up socket.close() context.term() 程序提供了 scatchsignals()，它可以捕获 Ctrl-C（SIGINT）和SIGTERM信号。当这些信号到达时，scatchsignals()处理程序将设置全局变量sinterrupted。由于信号处理程序，您的应用程序不会自动终止。相反，您有机会清理并优雅地退出。现在，您必须明确检查中断并正确处理它。通过在主代码的开头调用 scatch_signals()（从interrupt.c复制此内容）来实现此目的。这将设置信号处理。信号将如下影响ZeroMQ调用：

Multithreaded Hello World server

Author: Guillaume Aubert (gaubert)

"""import timeimport threadingimport zmq

def worker_routine(worker_url: str, context: zmq.Context = None): """Worker routine""" context = context or zmq.Context.instance()

# Socket to talk to dispatcher socket = context.socket(zmq.REP) socket.connect(worker_url)

while True: string = socket.recv() print(f"Received request: [ {string} ]")

# Do some 'work' time.sleep(1)

# Send reply back to client socket.send(b"World")

def main(): """Server routine"""

url_worker = "inproc://workers" url_client = "tcp://*:5555"

# Prepare our context and sockets context = zmq.Context.instance()

# Socket to talk to clients clients = context.socket(zmq.ROUTER) clients.bind(url_client)

# Socket to talk to workers workers = context.socket(zmq.DEALER) workers.bind(url_worker)

# Launch pool of worker threads for i in range(5): thread = threading.Thread(target=worker_routine, args=(url_worker,)) thread.daemon = True thread.start()

zmq.proxy(clients, workers)

# We never get here but clean up anyhow clients.close() workers.close() context.term()

if __name__ == "__main__": main()

所有的代码现在应该对您来说都是可以认识的。它的工作原理：

请注意，在大多数编程语言中创建线程都不是可移植的。POSIX库是pthreads，但在Windows上，您必须使用不同的API。在我们的示例中，pthreadcreate调用启动了一个新线程，运行我们定义的workerroutine函数。我们将在第3章-高级请求-回复模式中看到如何使用可移植的API包装它。

在这里，“工作”只是一秒的暂停。我们可以在工人中做任何事情，包括与其他节点交流。这就是多线程服务器在ØMQ套接字和节点方面的外观。请注意请求-回复链如何是REQ-ROUTER-queue-DEALER-REP。

当您使用ZeroMQ开始制作多线程应用程序时，您将遇到如何协调线程的问题。虽然您可能被诱惑插入“sleep”语句，或使用信号量或互斥量等多线程技术，但您应该使用的唯一机制是ZeroMQ消息。记住醉汉和啤酒瓶的故事。

让我们制作三个线程，当他们准备好时相互发信号。在这个例子中，我们使用inproc传输的PAIR套接字：

mtrelay.py

"""

Multithreaded relay

Author: Guillaume Aubert (gaubert)

"""

import threadingimport zmq

def step1(context: zmq.Context = None): """Step 1""" context = context or zmq.Context.instance() # Signal downstream to step 2 sender = context.socket(zmq.PAIR) sender.connect("inproc://step2")

sender.send(b"")

def step2(context: zmq.Context = None): """Step 2""" context = context or zmq.Context.instance() # Bind to inproc: endpoint, then start upstream thread receiver = context.socket(zmq.PAIR) receiver.bind("inproc://step2")

thread = threading.Thread(target=step1) thread.start()

# Wait for signal msg = receiver.recv()

# Signal downstream to step 3 sender = context.socket(zmq.PAIR) sender.connect("inproc://step3") sender.send(b"")

def main(): """ server routine """ # Prepare our context and sockets context = zmq.Context.instance()

# Bind to inproc: endpoint, then start upstream thread receiver = context.socket(zmq.PAIR) receiver.bind("inproc://step3")

thread = threading.Thread(target=step2) thread.start()

# Wait for signal string = receiver.recv()

print("Test successful!")

receiver.close() context.term()

if __name__ == "__main__": main()

这是使用ZeroMQ进行多线程编程的经典模式：

请注意，使用此模式的多线程代码不能扩展到进程。如果使用inproc和socket pairs,，则您正在构建紧密绑定的应用程序，即线程结构上互相依赖的应用程序。当延迟非常关键时，请这样做。另一种设计模式是松散绑定的应用程序，其中线程具有自己的上下文，并通过ipc或tcp进行通信。您可以轻松将松散绑定的线程分解为单独的进程。

这是我们第一次展示使用PAIR套接字的示例。为什么使用PAIR？其他套接字组合可能看起来可行，但它们都有可能干扰信令的副作用：

线程和节点的第二个重要差异是，你通常有一定数量的线程，但节点数量更多变化。让我们从我们早期的场景（天气服务器和客户端）中取一个，并使用节点协调来确保订阅者在启动时不会丢失数据。 这是应用程序的工作方式：

# We wait for 10 subscribersSUBSCRIBERS_EXPECTED = 10

def main(): context = zmq.Context()

# Socket to talk to clients publisher = context.socket(zmq.PUB) # set SNDHWM, so we don't drop messages for slow subscribers publisher.sndhwm = 1100000 publisher.bind("tcp://*:5561")

# Socket to receive signals syncservice = context.socket(zmq.REP) syncservice.bind("tcp://*:5562")

# Get synchronization from subscribers subscribers = 0 while subscribers < SUBSCRIBERS_EXPECTED: # wait for synchronization request msg = syncservice.recv() # send synchronization reply syncservice.send(b'') subscribers += 1 print(f"+1 subscriber ({subscribers}/{SUBSCRIBERS_EXPECTED})")

# Now broadcast exactly 1M updates followed by END for i in range(1000000): publisher.send(b"Rhubarb")

publisher.send(b"END")

if __name__ == "__main__": main() 这里是订阅者：syncsub.py## Synchronized subscriber#import timeimport zmq

def main(): context = zmq.Context()

# First, connect our subscriber socket subscriber = context.socket(zmq.SUB) subscriber.connect("tcp://localhost:5561") subscriber.setsockopt(zmq.SUBSCRIBE, b'')

time.sleep(1)

# Second, synchronize with publisher syncclient = context.socket(zmq.REQ) syncclient.connect("tcp://localhost:5562")

# send a synchronization request syncclient.send(b'')

# wait for synchronization reply syncclient.recv()

# Third, get our updates and report how many we got nbr = 0 while True: msg = subscriber.recv() if msg == b"END": break nbr += 1

print(f"Received {nbr} updates")

if __name__ == "__main__": main() 这是一个启动十个订阅者并启动发布者的Bash shell脚本：echo "Starting subscribers..."for ((a=0; a

订阅执行前缀匹配。也就是说，它们查找“所有以XYZ开头的消息”。显而易见的问题是：如何分隔键和数据，以便前缀匹配不会意外地匹配到数据。最好的答案是使用信封，因为匹配不会跨帧边界。下面是发布-订阅信封在代码中的最简示例。该发布者发送两种类型的消息A和B。信封保存消息类型：psenvpub.py"""

Pubsub envelope publisher

Author: Guillaume Aubert (gaubert)

"""import timeimport zmq

def main(): """main method"""

# Prepare our context and publisher context = zmq.Context() publisher = context.socket(zmq.PUB) publisher.bind("tcp://*:5563")

while True: # Write two messages, each with an envelope and content publisher.send_multipart([b"A", b"We don't want to see this"]) publisher.send_multipart([b"B", b"We would like to see this"]) time.sleep(1)

# We never get here but clean up anyhow publisher.close() context.term()

if __name__ == "__main__": main() 订阅者仅想接收类型为 B 的消息：psenvsub.py"""

Pubsub envelope subscriber

Author: Guillaume Aubert (gaubert)

"""import zmq

def main(): """ main method """

# Prepare our context and publisher context = zmq.Context() subscriber = context.socket(zmq.SUB) subscriber.connect("tcp://localhost:5563") subscriber.setsockopt(zmq.SUBSCRIBE, b"B")

while True: # Read envelope with address [address, contents] = subscriber.recv_multipart() print(f"[{address}] {contents}")

# We never get here but clean up anyhow subscriber.close() context.term()

if __name__ == "__main__": main() 运行这两个程序时，订阅者应该会显示这个：[B] We would like to see this[B] We would like to see this[B] We would like to see this... 这个例子展示了订阅过滤器如何接受或者拒绝整个多帧消息（键加数据）。你不会收到部分多帧消息。如果你订阅了多个发布者并且想要知道他们的地址，以便你可以通过另一个套接字向他们发送数据（这是一个典型的用例），你可以创建一个三部分消息。

高水位标记 当你能够快速地在进程之间发送消息时，你很快就会发现内存是一种宝贵的资源，而且很容易被填满。某个进程中的几秒延迟可能会变成一个积压，除非你了解问题并采取预防措施，否则可能会导致服务器崩溃。 假设你有进程 A 高频地向进程 B 发送消息，而进程 B 正在处理这些消息。突然间 B 变得非常繁忙（垃圾回收、CPU 超载等等），在短时间内无法处理这些消息。这可能是一些重度垃圾回收需要几秒钟，或者可能更长，如果出现更严重的问题。进程 A 仍然试图疯狂地发送消息会发生什么？一些消息将停留在 B 的网络缓冲区中，一些将停留在以太网线上，一些将停留在 A 的网络缓冲区中。其余的消息将以应用程序背后 A 发送它们的速度快速积累在 A 的内存中。如果你不采取一些预防措施，A 可以很容易地耗尽内存并崩溃。 这是消息代理常见的、经典的问题。更让人痛苦的是，表面上看起来是 B 的问题，而 B 通常是用户编写的应用程序，A 对其没有控制权。 有什么解决方案吗？其中之一是将问题传递到上游。A 从其他地方获取消息。所以告诉那个进程：“停下！”等等。这被称为流量控制。这听起来很有道理，但如果你正在发送 Twitter 的推文呢？难道你要让全世界停止发推文，直到 B 准备好了吗？ 流量控制在某些情况下有效，但在某些情况下则无效。传输层无法告诉应用层“停止”，就像地铁系统无法告诉大型企业“请再工作半个小时。我太忙了”。消息传递的答案是对缓冲区的大小设置限制，然后在达到这些限制时采取一些明智的行动。在某些情况下（不适用于地铁系统），最佳策略是放弃消息。在其他情况下，最好的策略是等待。 ZeroMQ使用HWM（高水位标记）的概念来定义其内部管道的容量。每个与套接字相连的连接都有自己的管道，并且具有发送和/或接收的HWM，具体取决于套接字类型。有些套接字（PUB、PUSH）只有发送缓冲区。有些（SUB、PULL、REQ、REP）只有接收缓冲区。有些（DEALER、ROUTER、PAIR）同时具有发送和接收缓冲区。 在 ZeroMQ v2.x 中，默认情况下 HWM 是无限制的。这很简单，但对于高容量的发布者来说通常是致命的。在 ZeroMQ v3.x 中，默认值为 1000，这更加合理。如果您仍在使用 ZeroMQ v2.x，则应始终在套接字上设置 HWM，无论是设置为 1000 以匹配 ZeroMQ v3.x 还是根据您的消息大小和预期订阅者性能考虑另一个数字。 当套接字达到其HWM时，它将阻塞或丢弃数据，具体取决于套接字类型。PUB和ROUTER套接字将在达到其HWM时丢弃数据，而其他套接字类型将阻塞。在inproc传输中，发送方和接收方共享相同的缓冲区，因此真正的HWM是两侧设置的HWM的总和。 最后，HWM并不是精确的；虽然默认情况下可能会得到高达1000条消息，但由于libzmq实现队列的方式，实际缓冲区大小可能会低得多（只有一半）。如何解决丢失消息问题 随着你使用 ZeroMQ 构建应用程序，你会多次遇到这个问题：丢失你期望接收到的消息。我们制作了一张图表，介绍了最常见的导致消息丢失的原因。Figure 25 - Missing Message Problem Solver

以下是该图表的总结：