我有一個 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 zmqzmq.zmq_version() # '4.3.2'zmq.__version__ # '19.0.2'我使用的是 MacOS 10.13.6。我完全沒有主意了?；ヂ摼W，請幫忙！
1 回答
catspeake
TA貢獻1111條經驗 獲得超0個贊
一個有效的異步變體是
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 變體）明顯不同。
添加回答
舉報
0/150
提交
取消