python和asyncua中的异步队列

Question

我正在使用 python 3.8.7 中的 asyncio 和 asyncua 库，我想使用 asyncio.Queue 将数据从 async 函数传递到主线程。

但是，我遇到的问题是我的队列在第一个await 之后没有收到任何消息。

这是我的例子：

import threading
import time
import asyncio
from asyncua import Client
import aiohttp

async def main(queue1):
    queue1.put_nowait("main")
    client = Client(url='opc.tcp://localhost:4840/freeopcua/server/')
    #client = aiohttp.ClientSession()

    async with client:
        await queue1.put("in_async")
        queue1.put_nowait("in_async_2")

    queue1.put_nowait("after await")


def wrapper(queue1: asyncio.Queue):
    queue1.put_nowait("wrapper")
    asyncio.run(main(queue1))

if __name__ == '__main__':
    queue1 = asyncio.Queue()
    t = threading.Thread(target=wrapper, args=(queue1,))
    t.start()
    time.sleep(1)
    noEx = True
    while noEx:
        try:
            x = queue1.get_nowait()
            print(x)
        except asyncio.QueueEmpty:
            noEx = False

我明白了：

wrapper
main

如果我使用其他异步库（示例中为 aiohttp），那么一切都会按预期工作：

wrapper
main
in_async
in_async_2
after await

我已经验证 minimal asyncua server 在 opc.tcp://localhost:4840/freeopcua/server/ 服务器工作 - 我可以使用此示例中的代码获取数据，但队列似乎不起作用。我有什么遗漏吗？

Answer 1

这是我的一个猜测，但我认为你有一个竞争条件。这是主线程中的while循环：

while noEx:
    try:
        x = queue1.get_nowait()
        print(x)
    except asyncio.QueueEmpty:
        noEx = False

一旦有东西打印出来，循环就会尝试获取队列中的下一个东西。如果此时队列为空，while 循环将退出，并且不会再打印任何内容。

如果在您的辅助线程中建立连接的速度足够快，则队列将填充接下来的两条消息；但是如果那里有一点延迟，那么您的主 while 循环可能在消息有机会打印之前已经退出。我很确定在进入 async with: block 之前可能存在（并且未知）网络延迟。

我会尝试在 print(x) 语句之后插入一个显着的时间延迟，看看会发生什么（如果这解决了问题，您可以稍后改进代码）。或者更改 while 循环，使其永远运行，因为您始终可以使用 control-C 退出程序。