python - 如何在 pyzmq 中使用 inproc 传输？

Question

我已经设置了两个小脚本来模拟 pyzmq 的发布和订阅过程。但是，我无法使用 inproc 传输将消息发送到我的订户客户端。我可以很好地使用 tcp://127.0.0.1:8080，只是没有 inproc。

pub_server.py

import zmq
import random
import sys
import time

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("inproc://stream")

while True:
    socket.send_string("Hello")
    time.sleep(1)

sub_client.py

import sys
import zmq

# Socket to talk to server
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.setsockopt_string(zmq.SUBSCRIBE, '')
socket.connect("inproc://stream")

for x in range (5):
    string = socket.recv()
    print(string)

我怎样才能成功地改变我的代码，以便我能够在我的两个脚本之间使用 inproc 传输方法？

编辑:

我更新了我的代码以进一步反射(reflect)@larsks 的评论。我仍然没有收到我发布的字符串 - 我做错了什么？

import threading
import zmq

def pub():
    context = zmq.Context()
    sender = context.socket(zmq.PUB)
    sender.connect("inproc://hello")
    lock = threading.RLock()

    with lock:
        sender.send(b"")

def sub():
    context = zmq.Context()
    receiver = context.socket(zmq.SUB)
    receiver.bind("inproc://hello")

    pub()

    # Wait for signal
    string = receiver.recv()
    print(string)
    print("Test successful!")

    receiver.close()

if __name__ == "__main__":
    sub()

Answer 1

顾名思义，inproc 套接字只能在同一进程 中使用。如果您要重写您的客户端和服务器，以便在同一进程中有两个线程，您可以使用 inproc，但否则这种套接字类型根本不适合您正在做的事情。

documentation在这一点上非常清楚:

The in-process transport passes messages via memory directly between threads sharing a single ØMQ context.

更新

看看更新后的代码，第一个突出的问题是，虽然上面引用的文档说“...在共享单个 ØMQ 上下文的线程之间”，但您正在创建两个上下文在你的代码中。通常，您只会在程序中调用 zmq.Context()一次。

接下来，您永远不会为订阅者订阅任何消息，因此即使其他一切正常，您实际上也不会收到任何消息。

最后，您的代码将经历 slow joiner problem :

There is one more important thing to know about PUB-SUB sockets: you do not know precisely when a subscriber starts to get messages. Even if you start a subscriber, wait a while, and then start the publisher, the subscriber will always miss the first messages that the publisher sends. This is because as the subscriber connects to the publisher (something that takes a small but non-zero time), the publisher may already be sending messages out.

发布/订阅模型不适用于单个消息，也不是一种可靠的传输方式。

所以，总结一下:

• 在创建套接字之前，您需要创建一个共享的 ZMQ 上下文。
• 您可能希望您的发布者循环发布而不是发布一条消息。由于您正在尝试使用 inproc 套接字，因此您需要将两个函数放入单独的线程中。
• 您需要设置订阅过滤器才能接收消息。

有一个example using PAIR sockets在可能提供有用起点的 ZMQ 文档中。 PAIR 套接字旨在协调 inproc 套接字上的线程，与发布/订阅套接字不同，它们是双向的，不受“慢连接器”问题的影响。