所以，我构建了这个 ZeroMQ 管道架构的小例子，因为我很快就不得不做类似的事情，而且我正试图以正确的方式掌握管道概念。
https://gist.github.com/2765708
现在，这是完全异步的。控制器将一批任务分派给各个工作人员，这些工作人员依次向接收器发送消息。控制器和接收器是我架构的固定部分，而工作人员是动态的。那很完美。
但是，我想知道工人们何时完成了所有任务。在那个例子中，我确实知道消息的数量，但在现实生活中情况并非如此。我可能有 100 条消息或 10,000 条消息。那么，接收器或控制器如何知道工人何时完成了他们的任务呢？我必须执行一些操作，这些操作取决于发送给工人的工作的结论。
一个想法（免责声明：我对 0MQ 的经验很少！）：
在相反方向设置“确认”管道。由于控制器大概知道它已经分派了多少个任务给工人（例如它调用的次数send），它可以使用一个 PULL 套接字从每个工人接收一个小消息（例如一个整数），指示任务完成. 工作进程将其完成的结果分派给接收器，同时将确认发送回控制器。一旦控制器收集到正确数量的确认，它就可以在进行下一组工作之前进行任何必要的后处理。
您也可以将其向下游推送到水槽，但您需要先通知水槽预期的工作单元总数，然后再将它们分给工人。
我想扩展@bjlaub 的答案。一开始是评论，但我打字太多了。我同意确认的概念，但相信它可以起源于多个地方。
这种通信有多种方法，这完全取决于您在系统中所追求的行为。
首先，您可以在工作人员完成每个任务时发送消息，或者在接收器接收每个任务时从接收器发送消息。现在我没有解决套接字的类型，只是通信的行为。我相信从接收器发送它会更有效，因为您只需要一个连接回控制器而不是每个工作人员一个连接。接收器不需要知道总共有多少任务。只是它在收到每个结果后都会触发一条消息。控制器可以确定预期有多少，因为它是提交点，以及当它用尽提交时新的（计数）。
现在不管你是从worker还是sink发送消息，你都可以使用不同的socket类型。如果您希望控制器在所有工作完成之前完全阻塞，那么您可以让它成为推/拉，直到它收到 X 消息（消息内容可以是任何东西。它只是一个触发器）。
如果控制器希望在这些任务发生时能够做其他工作，这可能会受到限制。如果是这样，您可以使用 pub/sub，并让控制器订阅在任务完成时收到通知，并异步维护计数，直到总数得到满足。
最后，也许您希望控制器在您认为合适时向接收器询问状态。您可以为控制器设置一个 req/rep 模式来询问接收器它按需收到了多少请求。
我确信其中一种模式将满足您的特定需求。