python-3.x - 在后台同时运行 asyncio 任务

Question

很抱歉，因为我看到这个问题被问了很多，但看完所有问题后似乎没有一个能解决我的问题。我的代码看起来像这样

TDSession = TDClient()
TDSession.grab_refresh_token()
q = queue.Queue(10)
asyncio.run(listener.startStreaming(TDSession, q))
while True:
    message = q.get()
    print('oh shoot!')
    print(message)
    orderEntry.placeOrder(TDSession=TDSession)

我试过做 asyncio.create_task(listener.startStreaming(TDSession,q))，问题是我得到了

RuntimeError: no running event loop
sys:1: RuntimeWarning: coroutine 'startStreaming' was never awaited

这让我感到困惑，因为这似乎在这里有效 Can an asyncio event loop run in the background without suspending the Python interpreter?这就是我想要做的。

listener.startStreaming 函数看起来像这样

async def startStreaming(TDSession, q):
    streamingClient = TDSession.create_streaming_session()
    streamingClient.account_activity()
    await streamingClient.build_pipeline()
    while True:
        message = await streamingClient.start_pipeline()
        message = parseMessage(message)
        if message != None:
            print('putting message into q')
            print( dict(message) )
            q.put(message)

有没有办法让我可以在后台运行监听器？

编辑:我也试过这个，但它只运行 consumer 函数，而不是同时运行两者

TDSession.grab_refresh_token()
q = queue.Queue(10)
loop = asyncio.get_event_loop()
loop.create_task(listener.startStreaming(TDSession, q))
loop.create_task(consumer(TDSession, q))
loop.run_forever()

Answer 1

正如您所发现的，asyncio.run 函数会运行给定的协程，直到它完成。换句话说，它会等待 listener.startStreaming 返回的协程完成，然后再继续下一行。

使用 asyncio.create_task ，另一方面，要求调用者已经在 asyncio 循环中运行。来自文档:

The task is executed in the loop returned by get_running_loop(), RuntimeError is raised if there is no running loop in current thread.

您需要通过创建一个 async 函数，然后在该异步函数内调用 create_task 来将两者结合起来。

例如:

async def main():
  TDSession = TDClient()
  TDSession.grab_refresh_token()
  q = asyncio.Queue(10)
  streaming_task = asyncio.create_task(listener.startStreaming(TDSession, q))
  while True:
    message = await q.get()
    print('oh shoot!')
    print(message)
    orderEntry.placeOrder(TDSession=TDSession)

  await streaming_task  # If you want to wait for `startStreaming` to complete after the while loop

if __name__ == '__main__':
  asyncio.run(main())

编辑:根据您的评论，我意识到您想使用生产者-消费者模式，因此我还更新了上面的示例以使用 asyncio.Queue 而不是 queue.Queue，以便线程能够在两者之间跳转生产者(startStreaming)和消费者(while 循环)