如何检查通过自定义asyncio后台装饰器启动的所有任务是否已全部执行完成

你使用的background装饰器调用asyncio.run_in_executor后会返回asyncio.Future对象，代表异步执行的后台任务，只要收集所有任务的Future并等待全部完成即可实现需求，以下是两种常用实现方案：

方案1：手动收集任务Future（无侵入修改） #

不需要修改原有装饰器，只需调整main函数的调用逻辑，收集所有返回的Future后统一等待：

import asyncio

def main():
    # 存储所有后台任务对应的Future对象
    task_futures = []
    # 注意：不要用list作为变量名，这是Python内置类型关键字，会覆盖原有功能
    for i in your_list:
        fut = myfunction()
        task_futures.append(fut)
    # 等待所有任务执行完成，当前线程会阻塞直到全部任务结束
    asyncio.get_event_loop().run_until_complete(asyncio.gather(*task_futures))

方案2：装饰器内置任务追踪（适合多场景复用） #

如果你在多处调用该装饰器修饰的函数，不想每次手动收集Future，可以修改装饰器内置任务追踪逻辑，新增统一等待的辅助函数：

import asyncio
from weakref import WeakSet

# 用弱引用集合存储活跃任务，避免任务完成后无法回收导致内存泄漏
_active_background_futures = WeakSet()

def background(fn):
    def wrapped(*args, **kwargs):
        fut = asyncio.get_event_loop().run_in_executor(None, fn, *args, **kwargs)
        _active_background_futures.add(fut)
        # 任务完成后自动从集合中移除
        fut.add_done_callback(_active_background_futures.discard)
        return fut
    return wrapped

# 统一等待所有活跃后台任务完成的辅助函数
def wait_all_background_tasks():
    if not _active_background_futures:
        return
    asyncio.get_event_loop().run_until_complete(asyncio.gather(*_active_background_futures))

# 你的业务函数保持不变
@background
def myfunction():
    # 运行业务逻辑
    pass

修改后main函数可以直接调用辅助函数等待：

def main():
    for i in your_list:
        myfunction()
    # 直接等待所有后台任务完成，无需手动收集
    wait_all_background_tasks()

注意事项 #

• 如果你的程序本身是异步架构，main可以定义为async函数，直接await asyncio.gather(*task_futures)即可，不需要调用run_until_complete，入口用asyncio.run(main())启动即可
• 如果需要忽略任务抛出的异常，避免等待过程被中断，可以给asyncio.gather加上return_exceptions=True参数，异常会作为返回值返回
• Python 3.10+版本中asyncio.get_event_loop()在部分场景下会触发废弃警告，可以替换为asyncio.get_running_loop()，调用时需要确保存在运行中的事件循环
• 如果后台任务是CPU密集型，可以把run_in_executor的第一个参数替换为自己初始化的concurrent.futures.ProcessPoolExecutor实例，用多进程避免GIL性能限制

内容的提问来源于stack exchange，提问作者DigitalGreyHat