Python ZeroMQ PUSH/PULL 逻辑，将高水位标记设置为低端 puller 而不会丢失任何消息答案

【问题标题】：Python ZeroMQ PUSH/PULL logic, set high water mark to a low end puller without losing any messagePython ZeroMQ PUSH/PULL 逻辑，将高水位标记设置为低端 puller 而不会丢失任何消息
【发布时间】：2019-12-24 21:41:27
【问题描述】：

我正在使用简单的一对一 PUSH/PULL worker/server python 代码来发送和接收消息。

worker 使用PUSH 套接字向PULL 服务器发送消息。服务器处理单元不如worker强大，因此当发送太多消息时，服务器的RAM开始增长，直到系统杀死所有东西。

我尝试如下设置接收器高水位线：

worker_sock_in = ZMQ_CTXT.socket(zmq.PULL)
worker_sock_in.setsockopt(zmq.LINGER, 1000))
worker_sock_in.setsockopt(zmq.RCVTIMEO, 1000)) # detects if the link is broken
worker_sock_in.setsockopt(zmq.RCVHWM, 1000)
worker_sock_in_port = worker_sock_in.bind_to_random_port(listen_addr, port_start, port_end)

以下代码用于worker创建和发送消息：

sock_dest = ZMQ_CTXT.socket(zmq.PUSH)
sock_dest.setsockopt(zmq.LINGER, 1000))
sock_dest.setsockopt(zmq.SNDTIMEO, 1000)) # detects if the link is broken
sock_dest.setsockopt(zmq.SNDHWM, 0) # never block on sending msg
sock_dest.connect(sock_dest_address)
# sends a msg
self.sock_dest.send(msg, zmq.NOBLOCK)

它似乎解决了这个问题，但我的猜测是溢出消息只是被服务器丢弃了，这在我的情况下是不可接受的。

我的开发基于此thread，但我不确定是否理解答案的附加信息部分。

所以问题是，在 noblock 推/拉 zeromq 套接字上达到的 HWM 的真实行为是什么？有没有办法让推拉基础设施保证所有发送的消息都将由拉套接字接收而不会膨胀其内存或阻止发件人？

【问题讨论】：

这里是一个类似的question，这里是一个answer，但它并不能保证所有发送的消息（这种方式有错过发送）。因此，在您的情况下，消息代理可能是一个更好的选择，而不是像 ZMQ 这样的无代理。

标签： python zeromq distributed-system pyzmq low-latency

【解决方案1】：

Q : 有没有一种方法来拥有一个推拉基础设施保证所有发送的消息都将被拉取接收套接字不会膨胀其内存或阻塞发送者？

方法？是的，有：

内置的 ZeroWarranty（涵盖作为原始文件的 1:1 位副本传送或根本不传送的消息）将需要扩展 - 通过应用程序级协议（涵盖重新- 发送那些未交付的，直到确认）或将您的基础设施转移到使用但特定的有保证的交付协议，这将有助于满足这个高于标准的要求 - 使用 norm:// 传输类扩展并移动范式，在 PUSH/PULL 仍不处于 RTO 状态的情况下，将其转换为 PUB/SUB, XPUB/XSUB Scalable Formal Communication Pattern Archetype。

_{libzmq 中提供了一个新的传输选项。 "norm_engine.hpp" 和 "norm_engine.cpp" 文件为 ZeroMQ 提供了面向 NACK 的可靠多播 (NORM) 传输协议选项的实现。 NORM 是 RFC 5740 和支持文档中指定的 IETF 开放标准协议。海军研究实验室 (NRL) 提供了一个开源参考实现，托管在 http://www.nrl.navy.mil/itd/ncs/products/norm。

NORM 支持通过 IP 多播进行可靠的数据传输，但也支持单播（点对点）数据传输。 NORM 在用户数据报协议 (UDP) 之上运行，并通过基于 NACK 的自动重复请求 (ARQ) 支持可靠性，该请求使用数据包擦除编码进行非常有效的组通信。 NORM 还提供自动 TCP 友好的拥塞控制和支持端到端流量控制的机制。 NRL NORM 实现还可以配置为提供基本的类似 UDP 的尽力而为传输服务（没有接收器反馈），并且可以通过在传输中添加一些应用程序可设置的主动前向纠错 (FEC) 数据包来增强这一点。即，默认情况下，NORM 仅发送“反应性”FEC 修复数据包以响应 NACK，但也可以配置为主动发送添加的修复数据包以获得一定程度的可靠性，而无需接收器的任何反馈。除了它的 TCP 友好拥塞控制，NORM 也可以配置为固定速率操作，并且 NRL 实现支持一些额外的自动拥塞控制选项，适用于容易误码的无线通信环境。虽然其可靠的 ARQ 操作主要基于 NACK（检测到数据包丢失时的否定确认），但它还支持来自接收器的可选肯定确认 (ACK)，可用于传送确认和显式流量控制。}

膨胀内存要求有两种前进方式：一种 - 显式控制用于.send()-er，而不是淹没.send()-er 侧@ 987654330@-instance 的资源 (RAM) 即在受限资源限制内（主要是防止任何泛滥/丢弃的消息发生），另一个 - 具有足够的 RAM 并正确配置 Context()-instance，让所有数据流过。

Q：HWM达到在 noblock push/pull zeromq 套接字上的真实行为是什么？

首先，让我们揭开它的神秘面纱。 ZMQ_NOBLOCK-directive 指向本地，.send()-side Context() 立即将对 .send() 方法的调用返回给调用者，即不阻塞调用代码执行（消息有效负载用于在本地 ZeroMQ Context()-instance 内部进行进一步处理，无论其内部状态如何 - 经典的非阻塞代码设计）。

ZMQ_SNDHWM 相反指示Context()-instance，如何设置此套接字的阈值以及上述PUSH/PULL.send()-er 情况：

_{高水位线是对最大未完成消息数量的硬性限制，ØMQ 应在内存中为指定套接字正在与之通信的任何单个对等方排队。零值意味着没有限制。

如果已达到此限制，则套接字将进入异常状态，并且根据套接字类型，ØMQ 应采取适当的措施，例如阻止或丢弃已发送的消息。请参阅zmq_socket(3) 中的各个套接字描述，了解对每种套接字类型采取的确切操作的详细信息。

ØMQ 不保证套接字会接受尽可能多的ZMQ_SNDHWM 消息，实际限制可能会降低多达 60-70%，具体取决于套接字上的消息流。}

同时使用 ZMQ_XPUB_NODROP 可能有助于 norm://-transport-class 用例。

另外请注意，默认情况下，ZMQ_PUSH-sockets 的 API 会确认：

_{当ZMQ_PUSH套接字由于所有下游节点都达到高水位线而进入静音状态，或者如果根本没有下游节点，那么任何zmq_send(3)对套接字的操作应阻塞直到静音状态结束或至少有一个下游节点可供发送；消息不会被丢弃。}

对于表现不佳的嫌疑人（PULL-side），还可以使用 .getsockopt( ZMQ_RCVBUF )-方法在 O/S 端测试适当大小的设置，并适当调整大小，足够大.setsockopt( ZMQ_RCVBUF )，根据需要：

_{ZMQ_RCVBUF 选项应将套接字的底层内核接收缓冲区大小设置为以字节为单位的指定大小。值 -1 表示保持操作系统默认值不变。有关详细信息，请参阅SO_RCVBUF 套接字选项的操作系统文档。}

如果以上没有任何帮助，可以使用 zmq_socket_monitor 服务将自我诊断元平面注入到 ZeroMQ 基础架构中，并完全控制通常在外部发生的情况查看应用程序代码（根据需要反映内部 API 状态和转换）。

决定权在你。

【讨论】：

说messages are not discarded当ZMQ_PUSH进入静音状态时，配合NOBLOCK参数是这样吗？我会研究一下听起来很有希望的 NORM
NOBLOCK 指令与调用返回到 .recv() 调用方的方式有关，与 Socket 端点的内部状态无关（最好重新阅读API 文档，关于处理间接 errno 变量以反映内部 FSA 状态）

【解决方案2】：

我建议您在中间（发送者和接收者之间）添加一个代理，它将在给定时间内保存发送的消息。您将有义务制作将保存消息的代码逻辑，并在服务器未收到特定消息时收到通知。 0mq 不提供保存或恢复丢失消息的方法。

【讨论】：