我有一個(gè) XPUB/XSUB 設(shè)備和多個(gè)模擬發(fā)布者在一個(gè)進(jìn)程中運(yùn)行。在一個(gè)單獨(dú)的過程中，我想連接訂閱者并將收到的消息打印到終端。下面我將展示一個(gè)簡單函數(shù)的兩個(gè)變體來實(shí)現(xiàn)這一點(diǎn)。我將這些函數(shù)包裝為命令行實(shí)用程序。我的問題是 asyncio 變體永遠(yuǎn)不會(huì)接收消息。另一方面，非異步變體工作得很好。我已經(jīng)測試了 ipc 和 tcp 傳輸?shù)乃星闆r。在我的測試中，發(fā)布過程從未改變，除非我重新啟動(dòng)它以更改傳輸。這些消息是短字符串，大約每秒發(fā)布一次，因此我們不考慮性能問題。用戶程序無限期地在線msg = await sock.receive_multipart()。在 XPUB/XSUB 設(shè)備中，我有顯示消息轉(zhuǎn)發(fā)的儀器sock.setsockopt(zmq.SUBSCRIBE, channel.encode())，與非異步變體連接時(shí)相同。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()常規(guī)阻塞變體（工作正常）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()對(duì)于這個(gè)特定的工具，不需要使用 asyncio。但是，我在代碼的其他地方也遇到了這個(gè)問題，異步接收永遠(yuǎn)不會(huì)收到。因此，我希望通過在這個(gè)簡單的案例中弄清楚它，我將了解一般情況下出了什么問題。我的版本是import zmqzmq.zmq_version() # '4.3.2'zmq.__version__ # '19.0.2'我使用的是 MacOS 10.13.6。我完全沒有主意了。互聯(lián)網(wǎng)，請幫忙！
1 回答
catspeake
TA貢獻(xiàn)1111條經(jīng)驗(yàn) 獲得超0個(gè)贊
一個(gè)有效的異步變體是
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())
我的結(jié)論是，當(dāng)使用 asyncio zmq 時(shí)，必須通過在等待套接字的事件循環(huán)上運(yùn)行的調(diào)用來創(chuàng)建套接字。盡管原始形式?jīng)]有對(duì)事件循環(huán)做任何花哨的事情，但套接字的事件循環(huán)似乎與asyncio.run. 我不知道為什么，我沒有用 pyzmq 提出問題，因?yàn)樗麄兊奈臋n顯示了這個(gè)答案中的用法，沒有評(píng)論。
編輯回應(yīng)評(píng)論：
asyncio.run總是創(chuàng)建一個(gè)新的事件循環(huán)，因此可能為傳遞到的協(xié)同例程之外實(shí)例化的套接字創(chuàng)建的循環(huán)asyncio.run（如原始問題中的 asyncio 變體）明顯不同。
添加回答
舉報(bào)
0/150
提交
取消