python-asyncio - 从在单独线程中运行的同步代码中将项目放入 asyncio.Queue 的最简单方法

Question

Python抱怨说 RuntimeWarning: coroutine 'Queue.put' was never awaited

我已经搜索并看到存在像Janus这样的库来解决这样的问题。但是在 3.8 上是否存在更好的方法？

更新。我能够使用 create_task 将项目放入队列中，但它要么阻塞，get要么put直到系统中发生其他一些异步事件才停止阻塞，即使现在队列中应该有一个项目不需要阻塞. 任何想法为什么会发生这种情况？它需要大约 10-20 秒才能自动解除阻塞，但如果我发送另一个事件，它会立即解除对前一个事件的阻塞，但当前事件会有延迟，除非我通过发送另一个事件。

Answer 1

您是create_task从运行事件循环的线程外部调用的。您应该asyncio.run_coroutine_threadsafe改用：

if result:
    # tell asyncio to enqueue the result
    fut = asyncio.run_coroutine_threadsafe(
        tasks.completed.put(result), loop)
    # wait for the result to be enqueued
    fut.result()

（您应该在主线程中检索循环并将其传递给线程。）

如果您的队列是无限的并且您不需要处理背压，您可以put_nowait使用以下命令调用call_soon_threadsafe：

if result:
    # tell asyncio to enqueue the result
    loop.call_soon_threadsafe(
        tasks.completed.put_nowait, result)

我能够使用 create_task 将项目放入队列中，但它要么阻塞 get 要么 put 直到系统中发生一些其他异步事件才停止阻塞，即使现在队列中应该有一个项目不需要它阻止。

这是因为loop.create_task它不是线程安全的，所以它不会正确地通知事件循环发生了什么事。