除非我使用绑定，否则 Pyzmq SUB 不会收到消息

Question

我有 3 个进程，我们称它们为 host 和 worker1 和 worker2。我希望 worker1 和 worker2 能够通过 PUB/SUB 套接字直接相互通信（host 间歇性插入），所以我有以下设置：

# host
socket = ctx.Socket(zmq.PUB)
socket.bind('ipc:///tmp/comms')

# worker1
socket = ctx.Socket(zmq.PUB)
socket.connect('ipc:///tmp/comms')
socket.send(b'worker1')

# worker2
socket = ctx.Socket(zmq.SUB)
socket.connect('ipc:///tmp/comms')
socket.setsockopt(zmq.SUBSCRIBE, b'worker1')
socket.recv()

到目前为止，此设置不起作用。 worker1 发送正常，但 worker2 似乎从未收到该消息。但是，如果我现在将设置更改为：

# host
socket = ctx.Socket(zmq.PUB)
socket.connect('ipc:///tmp/comms')

# worker1
socket = ctx.Socket(zmq.PUB)
socket.connect('ipc:///tmp/comms')
socket.connect(b'worker1')

# worker2
socket = ctx.Socket(zmq.SUB)
socket.bind('ipc:///tmp/comms')
socket.setsockopt(zmq.SUBSCRIBE, b'worker1')
socket.recv()

它工作得很好。但是，如果我也绑定到host，它会再次停止工作。

这是为什么？如果我现在有workerN 也需要订阅worker1 会发生什么，我如何从所有进程中绑定？这些绑定/连接语义是什么？ host 不是长寿进程，通过binding 做正确的事吗，如果是，为什么worker2 在connecting 时无法接收？

---

MWE：https://gist.github.com/ooblahman/f8f9724b9995b9646ebdb79d26afd68a

import zmq
import time
from multiprocessing import Process

addr = 'ipc:///tmp/test_zmq'

def worker1():
    ctx = zmq.Context()
    sock = ctx.socket(zmq.PUB)
    sock.connect(addr)
    while True:
        sock.send(b'worker1')
        print('worker1 sent')
        time.sleep(1)

def worker2():
    ctx = zmq.Context()
    sock = ctx.socket(zmq.SUB)
    sock.connect(addr) # Change this to bind
    sock.setsockopt(zmq.SUBSCRIBE, b'worker1')
    while True:
        sock.recv()
        print('worker2 heard')

def main():
    ctx = zmq.Context()
    sock = ctx.socket(zmq.PUB)
    sock.bind(addr) # Change this to connect (or remove entirely)

    p1 = Process(target=worker1)
    p2 = Process(target=worker2)
    p1.start()
    p2.start()

    p1.join()

if __name__ == '__main__':
    main()

Answer 1

Q：“host，作为一个长期存在的过程，通过binding 做正确的事情，难道不是...？”

嗯，没有人能说出来。
没有任何此类说法的支持证据（“是 long-lived process" )，对于任何此类扣除的假设 ("_{... 如果是这样，}why 是 worker2 失败连接时接收？")

我们可以告诉的是，
worker2 可能会尝试使用任何一种方法，无论是{ .bind() | .connect() }，成功，但仍然从来没有得到任何保证（一个美妙的零之禅）无论它是否会收到POSACK。这取决于许多其他因素，distributed-computing 系统中的因素更多：

• 如果/何时/如何管理TOPIC-list（使用和调用.setsockopt( zmq.SUBSCRIBE, ... ) 方法的顺序），
• 如果PUB-side 代理确实发送过任何此类消息，这些消息实际上与任何积极订阅的主题匹配，
• 如果默认和/或配置的资源实际上都处于一种状态，这足以接受任何此类潜在可交付消息，用于.send()-er 端的传输，具有来自发送方@的所有必要传输方式987654332@-instance 到预期接收方Context()-instance 的手中，最后，通过.recv()-方法在任何预期接收方的本地交付，
  所以总体来说确实不好说。您的代码存在几个主要冲突：

worker1，在上面的第一个示例中，在 PUB 方面的角色实际上永远不会提供任何信息。

---

问：“这是为什么？”

要求PUB 设置到另一个PUB 的传输链路不是合法的ZeroMQ 正式通信原型模式- 对.connect() 方法的调用，负责要求这样的设置，应该得到检查返回一个错误，剩下的故事就像观察两个人，一个PUB-in-host和一个PUB-in-worker1，他们都试图尽可能多地向他们的麦克风，但从来没有人听过扬声器，所以从来没有听过其他人的声音。

可行，但不实用，是吗？

---

Q：“如果我现在有workerN，它也需要订阅worker1，我该如何绑定所有进程？”

我们不需要.bind()，除非我们愿意。

如果workerN 想订阅worker1，它可能是.bind()，但worker1 不知道出现了任何新的workerN（它可能想订阅.connect()），是吗？ ?因此，.connect() 通常用于动态变化的SUB-side N-ary 队列1:N-setup（实际上是M:N，原则上）-在已知传输上具有PUB-side -class（es ...是的，可以有多个AccessPoints使用多个，...是的，多个传输类-都可以从同一个PUB-agent用于与一个通信或许多远程SUB-agents - 不是很酷吗？） - 因为.bind()-(s) 为 N-agents 的动态群组建立了一个已知的访问点在未来某个未知的时间到.connect()。