如何解决为什么pyzmq订阅者与asyncio的行为有所不同?
我有一个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。但是,我也在代码的其他地方也遇到了这个问题,而异步接收却从未收到。因此,我希望通过在这种简单情况下将其清除,以了解总体上出了什么问题。
我的版本是
import zmq
zmq.zmq_version() # '4.3.2'
zmq.__version__ # '19.0.2'
我在MacOS 10.13.6上。
我完全没有主意。互联网,请帮忙!
解决方法
一个有效的异步变体是
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
总是创建一个新的事件循环,因此,显然是为在传递给asyncio.run
的协同例程之外实例化的套接字创建的循环(如原始问题中的asyncio变体)不同。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。