关于同步 ZeroMQ 管道架构的问题答案

【问题标题】：Questions on synchronous ZeroMQ pipeline architecture关于同步 ZeroMQ 管道架构的问题
【发布时间】：2012-05-22 06:05:49
【问题描述】：

所以，我构建了这个 ZeroMQ 管道架构的小例子，因为我很快就不得不做类似的事情，并且我正试图以正确的方式掌握管道概念。

https://gist.github.com/2765708

现在，这是完全异步的。控制器将一批任务分派给各个工作人员，这些工作人员依次向接收器发送消息。控制器和接收器是我架构的固定部分，而工作人员是动态的。那很完美。

但是，我想知道工人何时完成所有任务。在那个例子中，我确实知道消息的数量，但在现实生活中情况并非如此。我可能有 100 条消息或 10,000 条消息。那么，接收器或控制器如何知道工人何时完成了他们的任务呢？我必须执行一些操作，这些操作取决于发送给工人的工作的结论。

【问题讨论】：

这就是信号量的好处。它需要是一个分布式信号量，但“A”将增加信号量，“B”将减少它。如果信号量计数超过 X，则“A”会等到它小于 X 时才发布更多作业。当然'A'不必等待，它可以用来知道还有多少任务需要完成。如果您使用水槽，这将解决您的问题。

标签： queue message-queue zeromq

【解决方案1】：

我想扩展@bjlaub 的答案。一开始是评论，但我打字太多了。我同意确认的概念，但相信它可以起源于多个地方。

这种通信有多种方法，这完全取决于您在系统中所追求的行为。

首先，您可以在工作人员完成每个任务时发送消息，或者在接收器接收每个任务时从接收器发送消息。现在我没有解决套接字的类型，只是通信的行为。我相信从接收器发送它会更有效，因为您只需要一个连接回控制器而不是每个工作人员一个连接。接收器不需要知道总共有多少任务。只是它在收到每个结果后都会触发一条消息。控制器可以确定自从它是提交点以来预期有多少，以及当它用尽提交（计数）时是新的。

现在无论你是从worker还是sink发送消息，你都可以使用不同的socket类型。如果您希望控制器在所有工作完成之前完全阻塞，那么您可以让它成为推/拉，直到它收到 X 消息（消息内容可以是任何东西。它只是一个触发器）。

如果控制器希望在这些任务发生时能够做其他工作，这可能会受到限制。如果是这样，您可以使用 pub/sub，并让控制器订阅在任务完成时收到通知，并异步维护计数，直到总数得到满足。

最后，也许您希望控制器在您认为合适时向接收器询问状态。您可以为控制器设置一个 req/rep 模式来询问接收器它按需收到了多少请求。

我确信其中一种模式会满足您的特定需求。

【讨论】：

+1 - 没有考虑让接收器发送确认..这肯定会提高性能！还可能值得注意的是，您仍然可以在控制器中使用 PULL 套接字，如果在等待确认时需要做其他工作，则只需使用非阻塞 I/O（使用zmq_poll）。
@bjlaub：确实如此。无论如何，您都可以轮询套接字。这真的只是归结为您希望通信路径如何运作。
我并不需要对每项任务的反馈，我只需要知道所有员工何时完成了分配的任务。
@KlausS.：你是说这些信息不是你想要的吗？如果不让工作人员与控制器更紧密地绑定，就无法知道他们应该期待多少工作。只有控制器应该知道它总共有多少工作。
好的，所以我必须让控制器推送到接收器并让接收器计算工作人员的响应数量，看看它是否与控制器推送的数量相匹配？

【解决方案2】：

一个想法（免责声明：我对 0MQ 的经验非常少！）：

反向设置“确认”管道。由于控制器大概知道它已经分派了多少任务给工人（例如它调用send 的次数），它可以使用 PULL 套接字从每个工人接收一个小消息（例如一个整数），指示完成的任务。工作进程将其完成的结果分派给接收器，同时将确认发送回控制器。一旦控制器收集到正确数量的确认，它就可以在进行下一组工作之前进行任何必要的后处理。

您也可以将其向下游推送到接收器，但您需要通知接收器预期的工作单元总数之前将它们耕种给工人。

【讨论】：