PyZMQ 订阅者在使用请求套接字时不接收消息

Question

我正在使用 PyZMQ，但我遇到了一个相当奇怪的问题。我有两个类包装用于通信的套接字，MZLSubscriber 和MZLRequester。有一个包含它们的类，MZLLink。对于其中的每一个，我还测试了MZLSubscriberTest、MZLRequesterTest 和MZLinkTest。订阅者和请求者的测试正常工作，但MZLinkTest 没有收到任何订阅者消息。

下面似乎是相关代码，它们是 3 个类的构造函数以及 MZLSubscriber 的 run() 以及 MZLink 和 MZLSubscriber 的测试。

MZLink构造函数：

    # Host information
    self.host = host
    self.requestPort = requestPort
    self.subscriberPort = subscriberPort

    # Set up zmq context
    self.zmq_context = zmq.Context()

    # Set up subscriber and replier
    self.subscriber = MZLSubscriber(self.zmq_context, self.host, self.subscriberPort)
    self.requester = MZLRequester(self.zmq_context, self.host, self.requestPort)

    # Start subscriber
    self.subscriber.start()

MZLink测试：

# Constants
HOST = "localhost"
REQ_PORT = 5555
SUB_PORT = 5556

# Create Link
link = MIDASZMQLink(HOST, REQ_PORT, SUB_PORT)

link.close()

MZLRequester构造函数：

    # Initialize class member variables
    self.zmq_context = zmq_context
    self.host = host
    self.port = port

    # Set up reply socket
    self.socket = self.zmq_context.socket(zmq.REQ)

    # Connect socket
    self.socket.connect("tcp://{0}:{1}".format(self.host, self.port))

MZLSubscriber构造函数：

   # Initialize parent process
    Process.__init__(self)

    # Store zmq context and connection host/port
    self.zmq_context = zmq_context
    self.host = host
    self.port = port

    # Sockets. Don't set them up here because sockets are not thread safe
    self.socket = None

    # Queue to store data in
    # TODO: Make queue not overflow if events come in too quickly
    self.queue = Queue()

MZLSubscriber.run():

    # Parent call
    Process.run(self)

    # Set up subscriber socket in this thread
    self.socket = self.zmq_context.socket(zmq.SUB)
    self.socket.setsockopt_string(zmq.SUBSCRIBE, unicode())

    # Connect socket
    self.socket.connect("tcp://{0}:{1}".format(self.host, self.port))

    # While the thread is alive, poll for data to put into queue
    # Calling MZLSubscriber.stop() will automatically change this
    while self.is_alive():
        datum = self.socket.recv()
        self.queue.put(datum)

    # Disconnect and close socket.
    #FIXME: Doesn't get here because terminate() immediately stops the process
    self.socket.disconnect("tcp://{0}:{1}".format(self.host, self.port))
    self.socket.close()

MZLSubscriber测试：

# Host information
HOST = "localhost"
SUBSCRIBER_PORT = "5556"

# Set up zmq context
zmq_context = zmq.Context()

# Set up subscriber
subscriber = MZLSubscriber(zmq_context, HOST, SUBSCRIBER_PORT)

# Start subscriber
subscriber.start()

# Stop and join subscriber
subscriber.close()
subscriber.join()

订阅者线程似乎在datum = self.socket.recv() 阻塞，这让我认为这可能是套接字创建的一些问题。但是，它似乎仅在与订户合作时才有效。请求者似乎在这两种情况下都有效。另外，只要注释掉处理requester的两行代码，一切都会顺利进行。

我为代码墙道歉，但目前我什至无法缩小问题的来源。当我这样做时，我将删除不相关的代码。处理传入数据的测试代码已被删除。

澄清一下，我使用 Python 2.7 和 PyZMQ 14.3.1。

更新：似乎在主线程中运行 MZLSubscriber 而不是创建另一个 Process 会产生预期的结果，因此这似乎是某种线程安全。据我所知，zmq 上下文是线程安全的，但套接字不是。我认为这不会引起问题，因为我明确确保每个线程都有一个套接字。

更新 2： 如果在 MZLSubscriber 中设置套接字的调用从 run() 移动到 __init__，则套接字似乎接收到一小部分已发布消息，但确实有错误：

Process MZLSubscriber-1:
Traceback (most recent call last):
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/Users/user/Repos/midas-client/client/midasclient/mzlSubscriber.py", line 45, in run
    datum = self.socket.recv()
  File "socket.pyx", line 628, in zmq.backend.cython.socket.Socket.recv (zmq/backend/cython/socket.c:5616)
  File "socket.pyx", line 662, in zmq.backend.cython.socket.Socket.recv (zmq/backend/cython/socket.c:5436)
  File "socket.pyx", line 139, in zmq.backend.cython.socket._recv_copy (zmq/backend/cython/socket.c:1771)
  File "checkrc.pxd", line 21, in zmq.backend.cython.checkrc._check_rc (zmq/backend/cython/socket.c:6032)
ZMQError: Interrupted system call

我已经通过在MZLSubscriber.run() 中创建一个新的zmq.Context 来解决这个问题，尽管我认为如果 zmq 上下文是线程安全的，这应该没有必要。

Answer 1

看来我的问题是在不同的进程上使用多个 zmq 上下文。虽然 PyZMQ 文档声明 zmq 上下文是线程安全的，但我只能假设它意味着 Python 线程而不是进程。这很令人困惑，因为在 C 中，zmq 上下文是线程安全的，尽管运行方式类似于 Python multiprocessing.Process。

通过为每个进程创建一个 zmq 上下文解决了这个问题。