如何用asyncio实现ThreadPoolExecutor式的异步任务管理？

完全可以用asyncio实现你需要的所有功能，不需要结合ThreadPoolExecutor（除非你的异步函数内部包含阻塞的同步代码，这种情况才需要借助线程池，纯异步场景下没必要）。下面是对应实现方案：

核心实现代码 #

import asyncio

class CustomException(Exception):
    pass

async def do_job(a, b, c, d):
    if a < b:
        raise CustomException("A is smaller than b")
    # 模拟实际业务的异步耗时操作
    await asyncio.sleep(0.1)
    return True, "Work done"

async def main(max_concurrent=5):
    # 信号量控制最大并发数
    semaphore = asyncio.Semaphore(max_concurrent)
    # 维护任务与对应参数的映射，方便任务完成后关联参数
    task_map = {}

    # 初始化100个任务参数
    initial_params = [(i, 5, "fixed_c", "fixed_d") for i in range(100)]
    
    # 提交初始任务
    for a, b, c, d in initial_params:
        # 用信号量包装任务，确保并发数不超限
        async def wrapped_task(a=a, b=b, c=c, d=d):
            async with semaphore:
                return await do_job(a, b, c, d)
        task = asyncio.create_task(wrapped_task())
        task_map[task] = (a, b, c, d)
    
    # 循环处理完成的任务，同时支持动态添加新任务
    while task_map:
        # 等待任意一个任务完成
        done, _ = await asyncio.wait(task_map.keys(), return_when=asyncio.FIRST_COMPLETED)
        for completed_task in done:
            params = task_map.pop(completed_task)
            try:
                result = completed_task.result()
                print(f"任务参数{params}执行成功，结果：{result}")
                
                # 示例：根据条件动态添加新任务
                a, _, _, _ = params
                if a % 10 == 0:
                    new_a = a + 100
                    new_task_params = (new_a, 5, "new_c", "new_d")
                    async def new_wrapped(a=new_a, b=5, c="new_c", d="new_d"):
                        async with semaphore:
                            return await do_job(a, b, c, d)
                    new_task = asyncio.create_task(new_wrapped())
                    task_map[new_task] = new_task_params
                    print(f"新增任务：参数{new_task_params}")
                    
            except CustomException as e:
                print(f"任务参数{params}触发自定义异常：{str(e)}")
            except Exception as e:
                print(f"任务参数{params}执行失败：{str(e)}")

asyncio.run(main())

关键特性说明 #

1. 并发限制：
   通过asyncio.Semaphore实现最大并发数控制，每个任务执行前会自动获取信号量，执行完成后释放，确保同时运行的任务不超过设定值。

2. 实时结果与参数关联：
   用task_map字典保存任务对象和对应的运行参数，任务完成后可以直接通过任务对象取出参数，和你之前用ThreadPoolExecutor时的futures字典逻辑完全一致。

3. 错误处理：
   通过completed_task.result()获取结果时，任务中抛出的异常会被重新抛出，你可以用try-except捕获自定义异常或通用异常，和线程池的错误处理逻辑一致。

4. 动态添加任务：
   在处理完成的任务时，随时可以创建新的异步任务并加入task_map，后续循环会自动处理这些新增任务，比线程池的动态任务提交更灵活自然。

与ThreadPoolExecutor的差异 #

• 异步协程比线程更轻量，上下文切换开销极低，适合IO密集型场景的大规模任务。
• 纯异步场景下不需要额外的线程池，避免了线程调度的开销。
• 如果你的do_job函数内部包含阻塞的同步代码（比如同步IO、CPU密集型计算），可以用loop.run_in_executor结合ThreadPoolExecutor来包装这部分逻辑，但纯异步代码不需要这种操作。

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