Python ZMQ 和多处理导致 zmq.error.ZMQError : Interrupted system call

Question

我有一个 Python 脚本，我在其中绑定(bind)了多个(例如 5 个)ZMQ 接收器套接字，如下所示:

receiver_1 = context.socket(zmq.PULL)
receiver_1.bind("tcp://*:5555")
...
receiver_5 = context.socket(zmq.PULL)
receiver_5.bind("tcp://*:5559")

receivers = [receiver_1, ..., receiver_5]

然后我启动一些 Google Compute Engine 实例，它们连接相应的发送方套接字。

我想并行地从这些套接字中提取数据，因此我尝试使用多处理池来实现。代码看起来像这样:

def recv_result(i):
    result_str = receivers[i].recv()
    return cPickle.loads(result_str)

pool = multiprocessing.Pool()
while True:
    results = pool.map(recv_result, [i for i in range(len(receivers))])
    # break when all results have been received
    ...

我在运行脚本时遇到的错误如下所示:

Traceback (most recent call last):
  ...
  File ...
    results = pool.map(recv_result, [i for i in range(len(receivers))])
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 227, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 528, in get
    raise self._value
zmq.error.ZMQError: Interrupted system call

我也尝试过使用 multiprocessing.Process 实现相同的功能，但我得到了本质上相同的错误，尽管方式更加困惑。

我想做的是更有效地从我的 GCE 实例接收所有结果，因为我发现这是我脚本中的瓶颈(在我当前的工作实现中，我只有一个接收器套接字，并且它连续地接收来自所有 GCE 实例的结果)。如果有人能指出我当前代码中的错误，或者任何关于实现我的目标的更好方法的建议，我们将不胜感激!

Answer 1

一些提示:

• 很好，你使用 ZeroMQ - 它可以为你做很多事情而无需编写太多代码
• 不要过度优化。您不会通过 ZeroMQ 通信的多处理/线程获得任何好处，它已经非常快并且能够交换大量消息
• 如果使用线程/多进程，永远不要共享 zmq 上下文，它必须是线程私有(private)的，否则它会中断。这可能是您出现异常的原因。
• 如果当前代码使用阻塞 zeromq 发送和接收，请将其更改为非阻塞。查看如何使用轮询。