python - asyncio 任务已被销毁，但它正在等待

Question

我正在开发一个示例程序，它以 block 的形式从数据源(csv 或 rdbms)中读取数据，进行一些转换并通过套接字将其发送到服务器。

但是因为 csv 非常大，出于测试目的，我想在几 block 之后中断阅读。不幸的是出了点问题，我不知道是什么以及如何解决它。可能我必须取消一些，但现在确定在哪里以及如何取消。我收到以下错误:

Task was destroyed but it is pending!
task: <Task pending coro=<<async_generator_athrow without __name__>()>>

示例代码为:

import asyncio
import json

async def readChunks():
  # this is basically a dummy alternative for reading csv in chunks
  df = [{"chunk_" + str(x) : [r for r in range(10)]} for x in range(10)]
  for chunk in df:
    await asyncio.sleep(0.001)
    yield chunk

async def send(row):
    j = json.dumps(row)
    print(f"to be sent: {j}")
    await asyncio.sleep(0.001)


async def main():
    i = 0
    async for chunk in readChunks():
        for k, v in chunk.items():
            await asyncio.gather(send({k:v}))
        i += 1
        if i > 5:
            break
        #print(f"item in main via async generator is {chunk}")
    

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Answer 1

许多 async 资源，例如生成器，需要在事件循环的帮助下进行清理。当 async for 循环通过 break 停止迭代异步生成器时，生成器仅由垃圾收集器清理。这意味着任务挂起(等待事件循环)但被销毁(由垃圾收集器)。

最直接的解决方法是显式关闭生成器:

async def main():
    i = 0
    aiter = readChunks()      # name iterator in order to ...
    try:
        async for chunk in aiter:
            ...
            i += 1
            if i > 5:
                break
    finally:
        await aiter.aclose()  # ... clean it up when done

---

可以使用 asyncstdlib 简化这些模式(免责声明:我维护这个库)。 asyncstdlib.islice允许在完全关闭生成器之前获取固定数量的项目:

import asyncstdlib as a

async def main():
    async for chunk in a.islice(readChunks(), 5):
        ...

如果 break 条件是动态的，scoping the iterator在任何情况下都保证清理:

import asyncstdlib as a

async def main():
    async with a.scoped_iter(readChunks()) as aiter:
        async for idx, chunk in a.enumerate(aiter):
            ...
            if idx >= 5:
                break