使用 asyncio.Queue() 时如何取消获取？

Question

我正在用 asyncio 编写一个客户端并使用 q.get() 来等待来自服务器的响应。当我收到来自服务器的响应时，我将其放入队列中。如果服务器连接丢失，我将不再执行这些放置操作，并且可能会有任意数量的 await q.get() 挂起。

我应该如何取消它们？我注意到，当我删除队列时，await 仍在等待。

Answer 1

这看起来像你想要做的吗？我认为你有两个选择：

如果您保留未完成的获取计数，那么当您完成队列时，您可以多次放置（无）吗？

或者如果 None 是有效响应，则保留未完成期货的列表并自己调用取消。

import asyncio

async def qget(q):
  try:
    x = await q.get()
    q.task_done()
    print("qget done ",x)
  except asyncio.CancelledError as e:
    print("qget cancel exception ",e)
  except Exception as e:
    print("qget exception ",e)

async def run():

  q = asyncio.Queue()
  futs = []
  futs.append( asyncio.ensure_future( qget(q) ) )
  futs.append( asyncio.ensure_future( qget(q) ) )
  num = 2

  await asyncio.sleep(0.1)

  # Keep the number of outstanding gets and put None for each one
  if 1:
    for x in range(num):
      q.put_nowait(None)

  # Or keep the futures in a list and cancel them
  if 0:
    for f in futs:
      f.cancel()

  await asyncio.sleep(1)
  print("run loop done")


asyncio.run(run())

如果您查看the python code for the queue，它确实保留了一个名为 _getters 的列表，但没有用于访问它的公共 api。