Python+ZeroMQ在PUSH/PULL mod中正确处理有限的数据流

Question

我试图用Python和ZMQ实现一个分布式PUSH/PULL（某种MapReduce）模型，如本文所述：http://taotetek.net/2011/02/02/python-multiprocessing-with-zeromq/。在本例中，结果管理器确切地知道要等待多少消息以及何时将“完成”状态发送给工人。在

假设我有一个长度未知的大但有限的数据流。在这种情况下，我不知道该停在哪里。最后，我试图从呼吸机发送“FINISHED”，而不是结果管理器，但是，当然，工作人员在处理过程中收到它（因为它是一个单独的通道），并立即死亡，因此大量数据丢失。在

否则，如果我使用相同的工作消息队列来发送“已完成”状态—它被第一个可用的工作线程捕获，而其他线程则挂起，这也是意料之中的。在

这里还有别的型号的吗？或者你能给我指出一些关于这个案例的最佳实践吗？在

Answer 1

Otherwise, if I use the same work_message queue to send "FINISHED" state - it's being captured by first available worker while others hang, that's also as expected.

你可以很容易地解决这个问题。在

1. 将"FINISH"从呼吸机发送到结果管理器PULL socket。在
2. 结果管理器接收"FINISH"，并通过PUB socket将此消息发布给所有工作人员。在
3. 所有工人在SUB sockets上接收"FINISH"消息并自杀。在

这里有来自ZMQ Guide的示例代码，如何将某事物从呼吸机发送到结果管理器，在devide and Convery设计模式中。在