为什么 pyzmq 订阅者的行为与 asyncio 不同？

Question

我在一个进程中有一个 XPUB/XSUB 设备和多个模拟发布者运行。在一个单独的过程中，我想连接一个订阅者并将收到的消息打印到终端。下面我将展示一个简单函数的两个变体来做到这一点。我将这些函数包装为命令行实用程序。

我的问题是 asyncio 变体从不接收消息。

另一方面，非异步变体工作得很好。我已经测试了 ipc 和 tcp 传输的所有情况。发布过程在我的测试中从未改变，除非我重新启动它以更改传输。消息是短字符串，大约每秒发布一次，因此我们不考虑性能问题。

订户程序无限期地位于 msg = await sock.receive_multipart() 行。在 XPUB/XSUB 设备中，我有显示 sock.setsockopt(zmq.SUBSCRIBE, channel.encode()) 消息转发的检测，与非异步变体连接时相同。

asyncio 变体（不工作，如上所述）

def subs(url, channel):
    import asyncio

    import zmq
    import zmq.asyncio

    ctx = zmq.asyncio.Context.instance()
    sock = ctx.socket(zmq.SUB)
    sock.connect(url)
    sock.setsockopt(zmq.SUBSCRIBE, channel.encode())

    async def task():
        while True:
            msg = await sock.recv_multipart()
            print(' | '.join(m.decode() for m in msg))

    try:
        asyncio.run(task())
    finally:
        sock.setsockopt(zmq.LINGER, 0)
        sock.close()

常规阻塞变体（工作正常）

def subs(url, channel):
    import zmq

    ctx = zmq.Context.instance()
    sock = ctx.socket(zmq.SUB)
    sock.connect(url)
    sock.setsockopt(zmq.SUBSCRIBE, channel.encode())

    def task():
        while True:
            msg = sock.recv_multipart()
            print(' | '.join(m.decode() for m in msg))

    try:
        task()
    finally:
        sock.setsockopt(zmq.LINGER, 0)
        sock.close()

对于这个特定的工具，不需要使用 asyncio。但是，我在我的代码的其他地方也遇到了这个问题，异步 recv 永远不会收到。所以我希望通过在这个简单的案例中清理它，我将了解一般情况下出了什么问题。

我的版本是

import zmq
zmq.zmq_version()  # '4.3.2'
zmq.__version__  # '19.0.2'

我使用的是 MacOS 10.13.6。

我完全没有想法。互联网，求助！

Answer 1

一个工作的异步变体是

def subs(url, channel):
    import asyncio

    import zmq
    import zmq.asyncio

    ctx = zmq.asyncio.Context.instance()

    async def task():
        sock = ctx.socket(zmq.SUB)
        sock.connect(url)
        sock.setsockopt(zmq.SUBSCRIBE, channel.encode())

        try:
            while True:
                msg = await sock.recv_multipart()
                print(' | '.join(m.decode() for m in msg))
        finally:
            sock.setsockopt(zmq.LINGER, 0)
            sock.close()

    asyncio.run(task())

我得出的结论是，当使用 asyncio zmq 时，必须在等待套接字的事件循环上调用运行创建套接字。尽管原始形式没有对事件循环做任何花哨的事情，但套接字似乎有一个不同于 asyncio.run 使用的事件循环。我不确定为什么，我没有打开 pyzmq 的问题，因为他们的文档显示了这个答案中的用法，没有评论。

编辑回复评论：

asyncio.run 总是创建一个新的事件循环，因此循环可能是为在 co-routine 之外实例化的套接字创建的，传递给 asyncio.run（如原始版本中的 asyncio 变体问题）明显不同。

为什么 pyzmq 订阅者的行为与 asyncio 不同？

Why does pyzmq subscriber behave differently with asyncio?

python

python-asyncio

zeromq

publish-subscribe

pyzmq