python - 从另一个线程中止 zeromq recv() 或 poll() - 立即且无需等待超时

Question

我在许多配置中使用 Python 和 C++ 中的 ZeroMQ，我想知道从另一个中止 recv() 或 poll() 的最优雅的方法是什么线程(例如，在受控程序终止的情况下，但如果您想在不需要终止套接字的情况下停止监听)。

对比this问题 我不仅想避免不定式等待，还想从 recv() 或 poll() 返回立即。

我知道我可以像这样提供一个 timeout 并中止 recv():

poller = zmq.Poller()
poller.register(socket, zmq.POLLIN)

while _running:
    if poller.poll(timeout=100) == []:
        # maybe handle unwanted timout here..
        continue

    handle_message(socket.recv())

这将无休止地轮询套接字，直到从另一个线程将 _running 设置为 False - 最多 100 毫秒后我就完成了。

但这并不好 - 我有一个繁忙的循环，并且很难通过这种方式处理真正的超时，这可能是不需要的行为的结果。我也必须等待超时，这在大多数情况下并不重要，但是......你知道我的意思。

当然我可以轮询一个额外的套接字以进行堕胎:

abort_socket = context.socket(zmq.SUB)
abort_socket.setsockopt(zmq.SUBSCRIBE, b"")
abort_socket.connect(<abort-publisher-endpoint>)

poller = zmq.Poller()
poller.register(socket, zmq.POLLIN)
poller.register(abort_socket, zmq.POLLIN)

while _running:
    poll_result = poller.poll(timeout=1000)
    if socket in poll_result:
        handle_message(socket.recv())
    elif abort_socket in poll_result:
        break
    else:
        # handle real timeout here
        pass

但是这种方式也有缺点:

• 有点冗长 - 在我触发中止的地方我必须创建一个发布者并使用它来中止接收者
• abort_socket 只能在一个线程中使用，所以我必须确保这一点

所以我的问题是:这是怎么做到的？

我能以某种方式只使用像 Python 的 threading.Event 或 s.th. 这样的东西吗？在其他语言中类似，而不是可以像这样传递给轮询器的中止套接字？:

def listener_thread_fn(event)

    poller = zmq.Poller()
    poller.register(socket, zmq.POLLIN)
    poller.register(event, zmq.POLLIN)

    while _running:
        poll_result = poller.poll(timeout=1000)
        if socket in poll_result:
            handle_message(socket.recv())
        elif event in poll_result:
            break
        else:
            # handle real timeout here
            pass

因此，您只需首先创建一个 theading.Event()，将其传递给 listener_thread_fn 并调用 event.set()从任何线程中止。

Answer 1

对于 Python 和 pyzmq，recv() 或 poll() 中断时会引发错误；所以你可以简单地在异常发生时捕获它。 recv() 的示例:

while True:
    try:
        request = server.recv()
    except zmq.ZMQError as e:
        if e.errno == errno.EINTR:
            print('Clean exit now!')
            break
        raise

您可以轻松修改该代码以改用 poll()(其过程相同)。另外请注意，您需要:

import errno