### 实现 Python 中耗时长的函数循环调用并限制并发数为 32 在 Python 中，可以使用 `asyncio` 结合 `concurrent.futures.ThreadPoolExecutor` 或者直接使用 `asyncio.Semaphore` 来实现异步后台执行耗时任务，并且可以设置最大的并发数量。以下是两种不同的实现方式： #### 方法一：基于 `asyncio` 和 `Semaphore` 这种方法适用于 IO 密集型任务，通过信号量 (`Semaphore`) 控制最大并发数。 ```python import asyncio import random # 创建一个信号量对象，限制最大并发数为 32 semaphore = asyncio.Semaphore(32) async def long_running_task(task_id): async with semaphore: # 使用信号量控制并发 print(f"Task {task_id} started") await asyncio.sleep(random.randint(1, 5)) # 模拟耗时操作 print(f"Task {task_id} finished") async def main(): tasks = [] for i in range(100): # 循环创建 100 个任务 task = asyncio.create_task(long_running_task(i)) tasks.append(task) await asyncio.gather(*tasks) # 等待所有任务完成 if __name__ == "__main__": asyncio.run(main()) ``` 此代码片段中，`asyncio.Semaphore(32)` 确保任何时刻最多有 32 个任务正在运行[^4]。如果尝试启动更多任务，则这些额外的任务将被挂起直到有资源可用。 --- #### 方法二：基于 `concurrent.futures.ThreadPoolExecutor` 对于计算密集型任务或者需要与现有同步代码集成的情况，可以选择使用 `ThreadPoolExecutor` 配合 `futures.as_completed()` 来管理任务队列。 ```python from concurrent.futures import ThreadPoolExecutor, as_completed import time import random def long_running_task(task_id): print(f"Task {task_id} started (Thread ID: {threading.get_ident()})") time.sleep(random.randint(1, 5)) # 模拟耗时操作 print(f"Task {task_id} finished") return f"Result of Task {task_id}" with ThreadPoolExecutor(max_workers=32) as executor: futures = [executor.submit(long_running_task, i) for i in range(100)] for future in as_completed(futures): result = future.result() print(result) ``` 在这个例子中，`ThreadPoolExecutor(max_workers=32)` 将线程池大小固定为 32，这意味着同一时间最多会有 32 个线程在工作[^1]。剩余的任务会在前序任务完成后逐步加入执行队列。 需要注意的是，由于 GIL 的存在，多线程更适合于 IO 密集型任务而非 CPU 密集型任务[^2]。如果是后者，建议改用 `ProcessPoolExecutor` 替代 `ThreadPoolExecutor`。 --- ### 对比分析 - **IO 密集型任务**：优先选择 `asyncio` 方案，因为它更轻量化且易于维护。 - **CPU 密集型任务**：应采用 `ProcessPoolExecutor`，因为其能有效规避 GIL 的影响[^3]。 无论哪种方案，都需要根据实际应用场景调整参数配置以及优化性能表现。 ---