线程池与进程池：concurrent.futures高效并发

一、引言在Python开发中你一定遇到过这样的场景需要爬取1000个网页用for循环一个一个请求慢到怀疑人生要处理100万行数据想用多线程加速但手动创建和管理1000个线程简直是灾难写了一个CPU密集型计算程序开了一堆线程却发现性能反而更差了。这些问题的本质是你没有用好并发编程的基础设施——线程池和进程池。concurrent.futures模块是Python官方提供的高层并发接口它用统一、简洁的API屏蔽了底层线程和进程的复杂管理让你能用几乎一样的代码实现多线程和多进程并发。python # 用线程池做I/O密集型任务 from concurrent.futures import ThreadPoolExecutor def fetch_url(url): return requests.get(url).status_code with ThreadPoolExecutor(max_workers10) as executor: results list(executor.map(fetch_url, urls))python # 用进程池做CPU密集型任务代码几乎一样 from concurrent.futures import ProcessPoolExecutor def cpu_task(n): return sum(i*i for i in range(n)) with ProcessPoolExecutor(max_workers4) as executor: results list(executor.map(cpu_task, [1000000] * 4))统一的API、简洁的代码、强大的功能——这就是concurrent.futures的魅力所在。本文将带你从零开始全面掌握这一高效并发工具。二、为什么需要线程池和进程池2.1 不使用线程池的痛点如果你手动创建和管理线程会遇到以下问题python # 手动创建线程的痛点 import threading def task(i): # 做一些事情 pass # 痛点1创建1000个线程每个线程占用约8MB内存 threads [threading.Thread(targettask, args(i,)) for i in range(1000)] for t in threads: t.start() for t in threads: t.join() # 内存占用~8GB系统直接卡死创建、销毁开销大频繁创建销毁线程会严重拖慢性能。管理复杂需要手动维护线程列表、等待所有线程完成。资源不可控无限制创建可能导致内存耗尽或系统崩溃。结果获取麻烦需要借助队列等通信机制才能收集任务返回值。2.2 线程池/进程池的核心思想线程池和进程池的核心思想非常简单预先创建一批可复用的工作线程/进程任务来了直接分配任务完成后回收复用。打个比方线程池就像一家餐厅的服务团队——不需要每个客人来了才去招一个服务员而是提前培养好一批服务员客人来了直接安排。这样既节省了招聘培训的时间又能灵活应对高峰期。进程池则更像一支特种部队——每个成员都是独立的有自己的装备和资源适合执行独立且复杂的任务。2.3 concurrent.futures的优势统一APIThreadPoolExecutor和ProcessPoolExecutor实现相同的Executor接口一行代码即可在两者间切换。自动管理池的大小、任务队列、资源清理全部自动处理。Future对象优雅地获取异步任务的状态、结果和异常。with语句支持自动等待任务完成并释放资源。三、核心概念Executor与Future3.1 Executor执行器抽象concurrent.futures.Executor是一个抽象基类定义了执行异步任务的核心接口。它的两个子类ThreadPoolExecutor和ProcessPoolExecutor都实现了这些方法方法作用submit(fn, *args, **kwargs)提交单个任务返回Future对象map(func, *iterables, timeoutNone)批量提交任务按顺序返回结果迭代器shutdown(waitTrue)关闭执行器释放资源3.2 Future异步任务的收据当你调用executor.submit()时它会立即返回一个Future对象。这就像你在餐厅点餐时拿到的小票——你不需要站在厨房门口等着而是可以先去做别的事等餐好了凭小票取餐。Future对象提供了以下核心方法python from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers2) as executor: future executor.submit(pow, 2, 10) print(future.done()) # False任务还没完成 print(future.result()) # 1024阻塞等待结果方法说明result(timeoutNone)获取任务结果阻塞直到完成done()判断任务是否完成cancel()尝试取消任务如果还没开始exception()获取任务抛出的异常add_done_callback(fn)任务完成时自动调用的回调函数四、ThreadPoolExecutor线程池详解4.1 基本用法python from concurrent.futures import ThreadPoolExecutor import time def worker(name, delay): print(f线程 {name} 开始工作) time.sleep(delay) print(f线程 {name} 完成) return f结果-{name} # 创建线程池最多同时运行3个线程 with ThreadPoolExecutor(max_workers3) as executor: # 提交任务返回Future对象非阻塞 future1 executor.submit(worker, A, 2) future2 executor.submit(worker, B, 1) # 获取结果阻塞等待 print(future1.result()) # 等待约2秒 print(future2.result()) # 结果: 结果-B使用with语句是最佳实践它确保代码块结束时自动调用shutdown(waitTrue)等待所有任务完成并释放资源。4.2 批量提交map方法如果你需要对一批数据应用同一个函数map方法是最简洁的选择python def square(x): return x * x with ThreadPoolExecutor(max_workers4) as executor: # map返回迭代器按提交顺序产出结果 results executor.map(square, [1, 2, 3, 4, 5]) print(list(results)) # [1, 4, 9, 16, 25]map方法也支持多个可迭代对象python def add(a, b): return a b with ThreadPoolExecutor() as executor: results executor.map(add, [1, 2, 3], [10, 20, 30]) print(list(results)) # [11, 22, 33]4.3 灵活调度submit as_completedmap有一个限制它必须等所有任务都完成才能开始返回结果。如果你希望谁先完成就返回谁的结果可以使用submit配合as_completedpython from concurrent.futures import ThreadPoolExecutor, as_completed import random def fetch_data(id): delay random.uniform(0.5, 2) time.sleep(delay) return f任务{id}完成耗时{delay:.2f}s with ThreadPoolExecutor(max_workers3) as executor: # 提交所有任务 futures [executor.submit(fetch_data, i) for i in range(5)] # 按完成顺序获取结果 for future in as_completed(futures): result future.result() print(result) # 输出示例顺序不定谁先完成谁先输出 # 任务2完成耗时0.62s # 任务0完成耗时0.89s # 任务4完成耗时1.23s # ...as_completed的核心优势在于实时响应任务完成后立即处理无需等待最慢的任务优雅容错可以在循环中用try/except捕获单个任务的异常不影响其他任务资源可控可以轻松添加超时控制4.4 设置超时无论是result()还是as_completed都可以设置超时python # 单个任务超时 future executor.submit(slow_task, 10) try: result future.result(timeout5) # 最多等5秒 except TimeoutError: print(任务超时) future.cancel() # as_completed也支持超时 for future in as_completed(futures, timeout10): result future.result()五、ProcessPoolExecutor进程池详解5.1 基本用法ProcessPoolExecutor的用法和ThreadPoolExecutor几乎完全相同但用途不同python from concurrent.futures import ProcessPoolExecutor def cpu_intensive_task(n): CPU密集型任务计算斐波那契数列 a, b 0, 1 for _ in range(n): a, b b, a b return a if __name__ __main__: with ProcessPoolExecutor(max_workers4) as executor: # 提交多个任务真正并行执行 futures [executor.submit(cpu_intensive_task, 1000000) for _ in range(4)] results [f.result() for f in futures]5.2 使用map批量提交python from concurrent.futures import ProcessPoolExecutor def process_data(data): return data ** 2 data * 2 if __name__ __main__: data list(range(10000)) with ProcessPoolExecutor(max_workers4) as executor: # 注意map返回的是迭代器按顺序产出结果 results executor.map(process_data, data) processed list(results)5.3 Windows系统的特殊注意事项在Windows上使用ProcessPoolExecutor时必须将主程序入口代码放在if __name__ __main__:块中。这是因为Windows没有fork()机制子进程会重新导入主模块如果没有这个保护会导致无限递归创建进程。python # 正确写法 from concurrent.futures import ProcessPoolExecutor def worker(x): return x * x if __name__ __main__: with ProcessPoolExecutor(max_workers4) as executor: results executor.map(worker, range(10)) print(list(results))5.4 进程间通信的限制由于每个进程有独立的内存空间传递给进程池的参数和返回值必须是可序列化的能被pickle模块处理。python # 不可以lambda函数无法pickle executor.submit(lambda x: x*2, 10) # 不可以在函数内部定义的类 class LocalClass: pass executor.submit(lambda x: x, LocalClass()) # 可以模块级别定义的函数和类 def my_func(x): return x * 2 class MyClass: pass executor.submit(my_func, MyClass())5.5 进程池的工作原理ProcessPoolExecutor底层基于multiprocessing模块构建但提供了更高层的抽象1. 初始化时根据max_workers参数启动指定数量的子进程这些进程进入等待状态。2. 提交任务时函数和参数被序列化通过队列发送给空闲工作进程。3. 执行任务时工作进程反序列化后执行函数将结果序列化传回主进程。4. 资源管理使用with语句或显式调用shutdown()时向所有工作进程发送退出信号。内部简化流程示意主进程 - 任务队列 - 工作进程 - 结果队列 - 主进程ProcessPoolExecutor还包含一个队列管理线程负责从结果队列中收集完成的任务结果并与对应的Future对象关联这样主进程就能通过future.result()获取结果了。六、线程池 vs 进程池全面对比6.1 核心区别对比维度ThreadPoolExecutorProcessPoolExecutor底层模块threadingmultiprocessing适用场景I/O密集型网络、文件、数据库CPU密集型计算、图像处理、加密GIL影响受GIL限制无法并行执行Python代码绕过GIL每个进程有独立解释器可真正并行内存占用低共享父进程内存约8MB/线程高每个进程独立内存约50MB/进程启动速度快微秒级慢毫秒级需复制内存数据共享容易共享内存需加锁困难需序列化或IPC异常隔离一个线程崩溃可能影响整个进程进程崩溃不影响其他进程最大并发量数千数十到数百6.2 池大小设置指南ThreadPoolExecutorI/O密集型可以设置较大的值如50-200取决于目标服务的承载能力默认线程池大小通常是min(32, os.cpu_count() 4)对多数I/O任务够用。ProcessPoolExecutor通常设置为os.cpu_count()或略高如1~2超过CPU核心数只会增加上下文切换开销没有性能提升。python import os from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor # 获取CPU核心数 cpu_count os.cpu_count() # 推荐设置 thread_pool_size min(64, cpu_count * 4) # I/O密集型可适当放大 process_pool_size cpu_count # CPU密集型不宜超过核心数七、实战案例7.1 案例一多线程爬虫场景批量下载100个网页内容python import requests import time from concurrent.futures import ThreadPoolExecutor, as_completed urls [https://httpbin.org/delay/0.5] * 100 # 模拟延迟响应 def fetch(url): response requests.get(url, timeout10) return response.status_code # 同步版本 start time.time() sync_results [fetch(url) for url in urls] print(f同步耗时: {time.time() - start:.2f}s) # 约50秒 # 线程池版本 start time.time() with ThreadPoolExecutor(max_workers20) as executor: futures {executor.submit(fetch, url): url for url in urls} for future in as_completed(futures): status future.result() print(f线程池耗时: {time.time() - start:.2f}s) # 约2.5秒提升20倍7.2 案例二多进程图像处理场景批量处理图片灰度转换、尺寸调整等CPU密集型操作python from PIL import Image from concurrent.futures import ProcessPoolExecutor import os def process_image(image_path): 单张图片处理转为灰度并缩放到50% with Image.open(image_path) as img: # CPU密集型操作 gray img.convert(L) resized gray.resize((img.width // 2, img.height // 2)) output_path fprocessed_{os.path.basename(image_path)} resized.save(output_path) return output_path if __name__ __main__: images [fimg_{i}.jpg for i in range(100)] with ProcessPoolExecutor(max_workersos.cpu_count()) as executor: results list(executor.map(process_image, images)) print(f已处理 {len(results)} 张图片)7.3 案例三混合架构线程池 进程池场景下载图片 → 压缩处理I/O CPU混合任务python from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor import requests import asyncio def download_image(url): I/O密集型下载图片 response requests.get(url) return response.content def compress_image(image_data): CPU密集型压缩图片模拟 # 模拟压缩处理 return len(image_data) // 2 def process_image_pipeline(url): 混合流水线 # 下载I/O img_data download_image(url) # 压缩CPU compressed compress_image(img_data) return compressed # 方案1直接混合不推荐会在线程池中跑CPU任务 # 方案2分别处理 def main(): urls [...] # 步骤1用线程池并发下载 with ThreadPoolExecutor(max_workers20) as downloader: downloaded list(downloader.map(download_image, urls)) # 步骤2用进程池并行压缩 with ProcessPoolExecutor(max_workersos.cpu_count()) as compressor: compressed list(compressor.map(compress_image, downloaded)) return compressed八、异常处理与资源管理8.1 Future中的异常处理线程池或进程池中抛出的异常不会立即在主线程中出现而是被封装在Future对象中直到调用result()时才重新抛出。python def divide(a, b): return a / b with ThreadPoolExecutor() as executor: future executor.submit(divide, 10, 0) # 异常不会在这里抛出 # 必须调用result()才能捕获 try: result future.result() except ZeroDivisionError as e: print(f捕获到异常: {e})8.2 三种异常处理模式模式适用场景代码示例submit result少量任务需要立即定位失败future.result()as_completed大量任务希望容错收集for f in as_completed(futures):map管道式处理要求整体一致性results executor.map(...)python # 模式2as_completed 容错收集 with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(risky_task, i) for i in range(10)] successes [] failures [] for future in as_completed(futures): try: result future.result() successes.append(result) except Exception as e: failures.append(str(e)) print(f成功: {len(successes)}, 失败: {len(failures)})8.3 资源清理shutdown和BrokenProcessPoolpython # 正确的资源管理 with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(task, i) for i in range(10)] # with块结束时自动调用shutdown(waitTrue) 如果进程池中的子进程意外崩溃后续提交会抛出BrokenProcessPool异常此时需要重建执行器[reference:26]。 python from concurrent.futures import ProcessPoolExecutor, BrokenProcessPool try: with ProcessPoolExecutor(max_workers2) as executor: # 执行任务... pass except BrokenProcessPool as e: print(f进程池已损坏需要重建: {e})九、性能实测9.1 CPU密集型任务对比python import time from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor def cpu_heavy(n): count 0 for i in range(n): count i * i return count def benchmark_cpu(): n 50_000_000 workers 4 # 单线程 start time.time() [cpu_heavy(n) for _ in range(workers)] single time.time() - start # 线程池受GIL影响基本无提升 start time.time() with ThreadPoolExecutor(max_workersworkers) as ex: list(ex.map(cpu_heavy, [n] * workers)) thread time.time() - start # 进程池真正并行 start time.time() with ProcessPoolExecutor(max_workersworkers) as ex: list(ex.map(cpu_heavy, [n] * workers)) process time.time() - start print(f单线程: {single:.2f}s) print(f线程池: {thread:.2f}s) print(f进程池: {process:.2f}s) # 结果 单线程: 3.2s 线程池: 3.4s甚至略慢线程切换锁竞争 进程池: 0.9s接近4倍加速9.2 I/O密集型任务对比python import time import requests from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor def io_task(delay): time.sleep(delay) # 模拟网络I/O return delay def benchmark_io(): n_tasks 50 delay 0.1 workers 20 # 单线程 start time.time() [io_task(delay) for _ in range(n_tasks)] single time.time() - start # 线程池 start time.time() with ThreadPoolExecutor(max_workersworkers) as ex: list(ex.map(io_task, [delay] * n_tasks)) thread time.time() - start # 进程池启动开销大略慢 start time.time() with ProcessPoolExecutor(max_workersworkers) as ex: list(ex.map(io_task, [delay] * n_tasks)) process time.time() - start print(f单线程: {single:.2f}s) print(f线程池: {thread:.2f}s) print(f进程池: {process:.2f}s) # 结果 单线程: 5.0s 线程池: 0.35s 进程池: 0.45s进程创建开销大稍慢十、常见问题与避坑指南问1线程池和进程池可以嵌套使用吗可以但需注意 在线程池中使用ProcessPoolExecutor是可行的在进程池中使用ThreadPoolExecutor也是可行的但不要过度嵌套导致资源竞争。问2executor.map和as_completed该怎么选场景推荐需要保持输入顺序map需要实时处理完成的任务as_completed需要容错单个失败不影响其他as_completed简单的批量映射map问3future.result()会阻塞怎么避免阻塞主线程如果不想阻塞可以使用回调函数python def callback(future): print(f任务完成结果: {future.result()}) future executor.submit(task, arg) future.add_done_callback(callback) # 主线程继续执行其他操作问4为什么我的线程池跑CPU任务没有加速因为GIL的存在Python的CPU密集型任务在线程池中无法真正并行。改用ProcessPoolExecutor即可。问5为什么进程池执行时报AttributeError: Cant pickle local object进程池需要序列化参数和函数确保函数定义在模块顶层不在函数内部、参数类型是可pickle的如基本类型、模块级类实例、不使用lambda表达式。十一、concurrent.futures vs 其他并发方案方案适用场景优势劣势concurrent.futures中等并发任务批量处理API统一简洁自动管理不适合海量短任务threading/multiprocessing需要精细控制灵活性高代码复杂易出错asyncio超高并发I/O万级连接内存占用极低性能优异需要async/await改造Celery分布式任务队列跨机器扩展重需额外基础设施python # 简单场景用concurrent.futures with ThreadPoolExecutor() as executor: results executor.map(func, data) # 高并发I/O场景用asyncio async def main(): async with aiohttp.ClientSession() as session: tasks [fetch(session, url) for url in urls] results await asyncio.gather(*tasks)十二、总结核心要点要点总结统一APIThreadPoolExecutor和ProcessPoolExecutor用法完全一样选型原则I/O密集型用线程池CPU密集型用进程池Future对象代表异步任务的状态、结果和异常资源管理用with语句自动管理避免资源泄漏异常处理必须调用result()才能捕获任务异常