ZeroMQ 工作者应该如何安全地“挂断”？

Question

我本周开始使用 ZeroMQ，当使用请求-响应模式时，我不确定如何让工作人员安全地“挂断”并关闭他的套接字，而不会丢失消息并导致发送该消息的客户永远得不到回应。想象一个用 Python 编写的工人看起来像这样：

import zmq
c = zmq.Context()
s = c.socket(zmq.REP)
s.connect('tcp://127.0.0.1:9999')
while i in range(8):
    s.recv()
    s.send('reply')
s.close()

我一直在做实验，发现127.0.0.1:9999 的套接字类型为zmq.REQ 的客户提出公平排队请求可能会不幸让公平排队算法在worker 已经完成了最后一个send()，但在它运行以下close() 方法之前。在这种情况下，似乎请求被工作进程中的ØMQ堆栈接收和缓冲，然后当close()抛出与套接字关联的所有内容时，请求就会丢失。

工人如何“安全地”分离——有没有办法发出“我不再需要消息”的信号，然后 (a) 循环在信号传输期间到达的任何最终消息，(b) 生成他们的回复，然后 (c) 执行 close() 并保证没有消息被丢弃？

编辑：我想我想进入的原始状态是“半关闭”状态，在这种状态下无法接收到更多请求——并且发送者会知道——但是在哪里返回路径仍处于打开状态，因此我可以检查传入缓冲区中是否有最后到达的消息，如果缓冲区中有一条消息，则对其进行响应。

编辑：为了回答一个好问题，更正了描述，使等待消息的数量成为复数，因为可能有许多连接在等待回复。

Answer 1

您似乎认为您正在尝试避免“简单”的竞争条件，例如 in

... = zmq_recv(fd);
do_something();
zmq_send(fd, answer);
/* Let's hope a new request does not arrive just now, please close it quickly! */
zmq_close(fd);

但我认为问题在于公平排队（循环）使事情变得更加困难：您的工作人员甚至可能已经有几个排队的请求。如果轮到它接收一个新请求，发件人不会等待您的工作人员空闲后再发送一个新请求，因此在您调用 zmq_send 时，其他请求可能已经在等待。

事实上，您可能选择了错误的数据方向。与其让请求池向您的工作人员发送请求（即使您不想接收新请求），您可能希望让您的工作人员从请求队列中获取新请求，处理它，然后发送答案。

当然，这意味着使用XREP/XREQ，但我认为这是值得的。

编辑：我写了some code实现the other direction来解释我的意思。

Answer 2

我认为问题在于您的消息传递架构是错误的。您的工作人员应该使用 REQ 套接字发送工作请求，这样只有一个工作在工作人员排队。然后，要确认工作完成，您可以使用另一个 REQ 请求，该请求兼作前一个作业的 ack 并请求一个新作业，或者您可以使用第二个控制套接字。

有些人使用 PUB/SUB 进行控制，这样每个工作人员都会发布确认信息，而主服务器会订阅它们。

您必须记住，使用 ZeroMQ 时有 0 个消息队列。一个都没有！只是消息在发送方或接收方中缓冲，具体取决于高水位标记和套接字类型等设置。如果您确实需要消息队列，那么您需要编写一个代理应用程序来处理它，或者简单地切换到所有通信都通过第三方代理进行的 AMQP。

Answer 3

我也一直在考虑这个问题。您可能希望实现一条 CLOSE 消息，通知客户工人即将离开。然后，您可以让工作人员在关闭之前耗尽一段时间。当然不理想，但可能可行。

Answer 4

在尽可能快地向工作人员发送请求与在工作崩溃或死亡的情况下获得可靠性之间存在利益冲突。 ZeroMQ 指南有一整节解释了这个可靠性问题的不同答案。读一下，会有很大帮助的。

tl;dr 工作人员可能/将会崩溃，并且客户端需要重新发送功能。该指南提供了多种语言的可重用代码。

Answer 5

最简单的解决方案不是让客户在等待回复时超时，然后在没有收到回复时重试吗？

Answer 6

尝试在调用关闭之前睡觉。这在 2.1 中已修复，但在 2.0 中尚未修复。