### Python 异步执行函数并限制并发数 为了实现一个函数的异步执行，同时限制最大并发数量为32，并将其挂起以便嵌套进复杂代码中，可以利用 `asyncio` 和 `concurrent.futures` 的功能。下面详细介绍如何设计这样的系统。 --- #### 1. 使用 `asyncio.Semaphore` 控制并发数 `asyncio.Semaphore` 是一种有效的工具，用于限制同时运行的任务数量。通过创建一个信号量对象，并在每个任务开始前获取该信号量，在任务结束后释放它，就可以轻松地控制并发度。 以下是一个基于 `Semaphore` 的示例代码： ```python import asyncio # 创建一个信号量，限制最多32个并发任务 semaphore = asyncio.Semaphore(32) async def bounded_task(task_id): async with semaphore: # 确保同一时间只有最多32个任务运行 print(f"Task {task_id} is running.") await asyncio.sleep(1) # 模拟耗时操作 print(f"Task {task_id} completed.") async def main(): tasks = [bounded_task(i) for i in range(100)] # 创建100个任务 await asyncio.gather(*tasks) # 并发运行所有任务 # 启动事件循环 asyncio.run(main()) ``` 在此代码中，`semaphore` 被用来确保任何时候都只会有不超过32个任务正在运行[^1]。 --- #### 2. 将任务挂起至后台 为了让任务能够在后台运行而不阻塞主线程或其他部分的代码逻辑，可以使用 `asyncio.create_task` 方法将协程提交给事件循环作为独立任务运行。这种方式可以让任务脱离当前上下文继续执行。 下面是改进版的例子，展示如何将多个任务放入后台运行： ```python import asyncio semaphore = asyncio.Semaphore(32) async def background_task(task_id): async with semaphore: print(f"Background Task {task_id} started.") await asyncio.sleep(2) print(f"Background Task {task_id} finished.") def schedule_background_tasks(num_tasks=100): loop = asyncio.get_event_loop() for i in range(num_tasks): loop.create_task(background_task(i)) # 提交到后台运行 async def complex_code_logic(): print("Starting complex code logic...") await asyncio.sleep(5) # 模拟一些其他工作 print("Complex code logic done!") async def main(): schedule_background_tasks() # 开始调度后台任务 await complex_code_logic() # 主线程继续做自己的事情 asyncio.run(main()) ``` 在这个例子中，`schedule_background_tasks` 函数负责安排所有的后台任务，而不会阻碍 `complex_code_logic` 的正常执行[^3]。 --- #### 3. 结合 `concurrent.futures.ThreadPoolExecutor` 虽然 `asyncio` 很适合处理 I/O 密集型任务，但对于 CPU 密集型任务或需要同步 API 的场景，`concurrent.futures.ThreadPoolExecutor` 更加适用。两者也可以结合起来使用，以充分利用各自的优点。 以下是如何结合两者的示例： ```python from concurrent.futures import ThreadPoolExecutor import asyncio executor = ThreadPoolExecutor(max_workers=8) def cpu_bound_task(task_id): print(f"Synchronous task {task_id} starting on thread pool.") from time import sleep sleep(2) # 模拟CPU密集计算 print(f"Synchronous task {task_id} finished.") async def run_cpu_bound_tasks_in_executor(): loop = asyncio.get_running_loop() futures = [] for i in range(32): # 发起32个任务 future = loop.run_in_executor(executor, cpu_bound_task, i) futures.append(future) await asyncio.gather(*futures) async def main(): await run_cpu_bound_tasks_in_executor() asyncio.run(main()) ``` 这里，`run_in_executor` 方法被用来把同步函数交给线程池去执行，从而避免阻塞事件循环[^2]。 --- #### 4. 总结与注意事项 - **并发限制**：通过 `asyncio.Semaphore` 可以方便地设定最大并发数。 - **后台运行**：借助 `asyncio.create_task` 让任务进入后台运行模式。 - **混合模型**：对于不同类型的负载（I/O vs CPU），可以选择适当的技术栈——`asyncio` 或者 `concurrent.futures`。 - **死锁预防**：注意在线程池中不要引入可能导致死锁的操作链[^4]。 --- ###