在 Python 中，如果你希望**并发地对一个列表（list）中的每个元素执行某个函数**（map 操作），然后**将结果汇总**（reduce 或 collect 操作），你可以使用 `concurrent.futures` 模块中的 `ThreadPoolExecutor` 或 `ProcessPoolExecutor`。前者适用于 I/O 密集型任务，后者适用于 CPU 密集型任务。 下面是一个使用 `ThreadPoolExecutor` 实现并发 map 的示例。 --- ### 示例代码 ```python import concurrent.futures def process_item(item): # 模拟一个耗时操作，比如网络请求、文件读写等 return item * item def concurrent_map(func, items): results = [] with concurrent.futures.ThreadPoolExecutor() as executor: future_to_item = {executor.submit(func, item): item for item in items} for future in concurrent.futures.as_completed(future_to_item): try: result = future.result() results.append(result) except Exception as exc: print(f"Generated an exception: {exc}") return results # 示例使用 items = [1, 2, 3, 4, 5] results = concurrent_map(process_item, items) print("结果汇总:", results) ``` --- ### 代码解释： 1. **`process_item` 函数**：这是你希望对每个元素执行的操作，这里只是一个简单的平方操作。 2. **`ThreadPoolExecutor`**：创建一个线程池，用于并发执行任务。 3. **`executor.submit(func, item)`**：提交任务到线程池中执行。 4. **`future_to_item`**：是一个字典，用于将 future 对象与对应的 item 关联起来，便于调试或错误处理。 5. **`concurrent.futures.as_completed()`**：按任务完成的顺序返回 future 对象。 6. **`future.result()`**：获取任务的返回结果，如果任务出错则抛出异常。 7. **`results.append(result)`**：将每个任务的结果收集到一个列表中，实现“汇聚”。 --- ### 使用 `ProcessPoolExecutor`（适用于 CPU 密集型任务） ```python with concurrent.futures.ProcessPoolExecutor() as executor: ... ``` 只需将 `ThreadPoolExecutor` 替换为 `ProcessPoolExecutor` 即可切换为多进程并发。 --- ### 使用 `map` 简化版本（更简洁） ```python def concurrent_map_simple(func, items): with concurrent.futures.ThreadPoolExecutor() as executor: return list(executor.map(func, items)) ``` 这个版本更简洁，但不能在出错时做细粒度处理。 --- ### 注意事项： - 如果你的函数是 CPU 密集型的（如大量计算），使用 `ProcessPoolExecutor` 更高效。 - 如果你的函数是 I/O 密集型的（如下载、读写文件），使用 `ThreadPoolExecutor` 更合适。 - 线程之间共享内存，进程之间默认不共享内存。 --- ###