### Python 实现非阻塞循环并发执行 为了实现一个非阻塞的循环，允许并发执行耗时函数而不限制于同步等待，并且限制最大并发数为32，可以使用 `asyncio` 和 `concurrent.futures` 结合的方式。以下是具体方法： #### 使用 `ThreadPoolExecutor` 或 `ProcessPoolExecutor` 可以通过 `concurrent.futures.ThreadPoolExecutor` 或者 `ProcessPoolExecutor` 来管理线程或进程池的最大并发数[^2]。对于 CPU 密集型任务，建议使用 `ProcessPoolExecutor` 以规避 GIL 的影响；而对于 I/O 密集型任务，则更适合使用 `ThreadPoolExecutor`。 #### 将 `concurrent.futures` 整合至 `asyncio` 通过 `asyncio` 中的 `loop.run_in_executor()` 方法，可以在异步框架中集成传统的线程/进程池功能。这使得我们可以定义一个协程来提交任务给线程池或进程池，并控制其并发度。 以下是一个完整的示例代码，展示如何在 Python 中实现这一需求： ```python import asyncio import concurrent.futures import time import random def slow_function(task_id): """模拟一个耗时的操作""" sleep_time = random.uniform(0.5, 2) time.sleep(sleep_time) return f"Task {task_id} completed in {sleep_time:.2f} seconds" async def main(): tasks = [] max_concurrency = 32 # 设置最大并发数 loop = asyncio.get_event_loop() # 创建 ThreadPoolExecutor 或 ProcessPoolExecutor with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrency) as executor: for i in range(100): # 假设有100个任务需要处理 # 提交任务到线程池并通过 run_in_executor 运行 coro = loop.run_in_executor(executor, slow_function, i) tasks.append(coro) # 收集所有任务的结果 results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if not isinstance(result, Exception): print(result) if __name__ == "__main__": start_time = time.time() asyncio.run(main()) elapsed_time = time.time() - start_time print(f"\nTotal execution time: {elapsed_time:.2f} seconds") ``` 此代码片段展示了如何创建一个具有固定大小的工作线程池（或进程池），并将一系列耗时任务分配其中。通过 `run_in_executor` 函数桥接了传统多线程或多进程模型与现代异步编程模型[^4]。 #### 关键点解析 - **最大并发数**：通过设定 `ThreadPoolExecutor` 或 `ProcessPoolExecutor` 的 `max_workers` 参数为 32，实现了对并发量的有效控制。 - **非阻塞性质**：借助 `asyncio.gather` 同时启动多个协程，这些协程会分别向线程池提交任务并立即返回，不会因单个任务未完成而阻碍后续任务的发起。 - **异常捕获**：设置了 `return_exceptions=True` 参数以便能够区分正常结果和错误情况下的行为[^1]。 --- ###