代码之家 › 专栏 › 技术社区 › Brandon Rhodes

ZeroMQ工作人员应该如何安全地“挂断”?

zeromq rpc message-queue concurrency python

Brandon Rhodes · 技术社区 · 14 年前

本周我开始使用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 提出公平排队请求的人可能不幸的是,在公平排队算法完成最后一个请求后,选择上面的工作人员。 send() 但在它运行之前 close() 方法。在这种情况下,请求似乎是由工作进程中的_MQ堆栈接收和缓冲的,当 关闭() 抛出与套接字相关的所有内容。

工人如何才能“安全地”分离?是否有任何方法发出“我不再需要信息”的信号,然后(a)循环传送信号期间到达的任何最终信息,(b)生成他们的回复,然后(c)执行 关闭() 保证没有信息被丢弃?

编辑: 我想我要进入的原始状态是“半关闭”状态,在这种状态下,不能再收到进一步的请求,发送者会知道“半关闭”,但返回路径仍然是打开的,这样我就可以检查我的传入缓冲区中最后一条到达的消息,并在缓冲区中有一条消息时对其作出响应。

编辑: 为了回答一个好问题,请更正描述,使等待消息的数目变为复数,因为可能有许多连接等待答复。

6 回复 | 直到 13 年前

Samuel Tardieu 14 年前

你似乎认为你在试图避免一种简单的种族状况,比如

... = 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 解释我的意思。

Michael Dillon 13 年前

我认为问题在于您的消息体系结构是错误的。您的工作人员应该使用REQ套接字发送工作请求,这样工作人员就只有一个作业排队。然后,为了确认工作的完成,您可以使用另一个REQ请求,该请求加倍作为前一个作业的ACK并请求新的REQ请求,或者您可以拥有第二个控制套接字。

有些人使用pub/sub进行控制,以便每个工人发布ack,而master订阅它们。

您必须记住,使用zeromq时,有0个消息队列。一点也没有!根据设置(如高水位线和插座类型),只在发送器或接收器中缓冲消息。如果您确实需要消息队列,那么您需要编写一个代理应用程序来处理这个问题,或者只需切换到AMQP,其中所有通信都通过第三方代理进行。

bneal 14 年前

我也在考虑这个问题。您可能希望实现一条关闭消息,通知客户工人将要离开。然后你可以让工人在关机前先排干一段时间。当然不理想,但可能可行。

Pieter Hintjens 13 年前

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

tl;dr工作者可以/将崩溃,客户机需要重新发送功能。该指南提供了多种语言的可重用代码。

-1

progrium 14 年前

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

-3

Trey Stout 14 年前

试着在呼叫结束前睡觉。这在2.1中是固定的,但在2.0中还没有固定。