使用 pyzmq 与子进程通信

Question

我正在尝试通过 ZeroMQ 套接字与我开始使用 multiprocessing.Process 的子进程通信。我知道存在与multiprocessing 模块中的子进程通信的解决方案，但我想最终与用 C++ 编写的共享库中的函数通信。废话不多说，代码如下：

import time
import zmq
import multiprocessing

def perform(nseconds, endpoint):
    context = zmq.Context()
    publisher = context.socket(zmq.PUB)
    publisher.connect(endpoint)

    for i in range(5):
        time.sleep(nseconds)
        publisher.send_string("{}".format(i))
    publisher.send_string(">>END")

if __name__ == "__main__":
    multiprocessing.freeze_support()
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.bind("tcp://*:*")
    socket.setsockopt_string(zmq.SUBSCRIBE, u"")
    endpoint = socket.getsockopt_string(zmq.LAST_ENDPOINT)

    print("Binding via {}".format(endpoint))

    t = multiprocessing.Process(target=perform, args=(1,endpoint))
    t.start()

    string = ""
    while not ">>END" in string:
        string = socket.recv_string()
        print(string)

    t.join()

此代码在 GNU/Linux 上运行良好，并具有预期的输出：

Binding via tcp://0.0.0.0:34149
0
1
2
3
4
>>END

但是在 Windows 上使用来自 Anaconda 的 Python 和安装了 conda install pyzmq 的 pyzmq 版本 16.0.2 运行它会崩溃并出现以下错误：

Binding via tcp://0.0.0.0:52019
Assertion failed: Can't assign requested address (bundled\zeromq\src\tcp_connect
er.cpp:341)

我该如何解决这个问题？还是我做错了？如果我做错了，为什么它依赖于平台？

Answer 1

删除socket.bind( "tcp://*:*" ) 中的通配符

首先，
这是非常特定于平台的，永远不应该依赖通配符扩展将如何处理生产中未知的生态系统细节。明确。

接下来，
这与分布式系统设计中简洁、最先进的资源管理实践完全相反，从架构的角度来看，几乎没有比在所有端口上直接使用.bind() 更糟糕的想法了。 987654325@ 可用地址。想象一下这会导致Context()-instance 内部的原因，以管理所有端点群及其相关资源池，并准备好嗅探潜在的传入连接请求（以防万一出现此类请求）。没有。

永远不要这样做。

---

性能sins and other aspects _{- ref. my criticism on Amdahl's Law v/s costs}

好吧，看过您的出版物后，我不会对此进行过多扩展，但必须补充一点，分布式计算领域需要在细节上花费大量精力——永远不要指望几个 SLOC 对 HPC 有任何好处计算。

最严重的罪过来自于改善某些过程的善意。如果代码衍生出一个流程（multiprocessing 的设计目的正是如此），没有多少人还会将这样的 SLOC 与实际成本联系起来，这些成本必须在离开之前支付代码获得了第一次开始处理的机会（完整的 python 执行环境的副本（以便从原始的本地 GIL 等中逃脱）——因此您的代码必须同时支付 [TIME] 和[SPACE] 对于大量内存到内存的传输，接下来您建议的代码会实例化另一个“远程”进程 Context()-引擎。虽然这看起来很聪明，您的代码必须再次支付所有费用 - 并且只需花费一些 SLOC 来处理。接下来，请始终使用 .setsockopt( zmq.LINGER, 0 )，以免您的资源无限阻塞优雅终止。虽然这似乎是一个“很好”的预防措施，但它是一个“必须做”的救生员，在寻找“这次出了什么问题？”之前，在一个又大又昂贵的 c 上计算基础设施...

许多其他性能调整应该在之前完成multiprocessing.Process( ... )，这超出了本文的范围。但值得掌握，绝对是在运行多次低效代码之前。