Python concurrent.futures 实战:用 ThreadPoolExecutor 并发处理 + as_completed 收结果

Python concurrent.futures 实战:用 ThreadPoolExecutor 并发处理 as_completed 收结果你手头有 200 个 URL 要抓,串行requests.get一个个跑,每个网络往返 300ms,光等待就是一分多钟,CPU 全程在打瞌睡。这类「一堆各自独立、主要在等 IO」的活儿,正是线程池的主场。但很多人第一次用ThreadPoolExecutor就踩坑:要么用map拿不到异常,要么用submit后不知道怎么优雅地边完成边收结果。这篇把这些坑一次讲清楚。朴素写法:串行,慢到肉眼可见先看最直白的版本,抓一批 URL 的状态码:importtimeimportrequestsdeffetch_status(url:str)-int:rrequests.get(url,timeout5)returnr.status_code urls[https://httpbin.org/delay/1]*10starttime.perf_counter()results[fetch_status(u)foruinurls]# 一个跑完才跑下一个print(results,f{time.perf_counter()-start:.1f}s)# [200, 200, ...] 10.x s —— 每个 delay 1 秒,10 个就是 10 秒10 个请求跑了 10 秒。但这些请求彼此毫无依赖,完全可以同时发出去。用 ThreadPoolExecutor.map 并发:能跑,但异常会「延迟爆炸」第一个改进是把列表推导换成线程池的map:fromconcurrent.futuresimportThreadPoolExecutorwithThreadPoolExecutor(max_workers10)aspool:resultslist(pool.map(fetch_status,urls))# 10 个请求几乎同时发出,总耗时约 1 秒max_workers10表示最多 10 个线程同时干活,with退出时会自动等所有任务结束再关池。速度从 10 秒降到 1 秒。但map有个隐蔽问题:它保持输入顺序,而且异常会在你迭代到那一项时才抛出。如果第 3 个 URL 挂了,前两个结果你都拿到了,迭代到第 3 个时才raise,这时候想知道「到底哪些成功了」就很别扭。而且map一旦某项抛异常,后面的结果你也拿不到了。正解:submit as_completed,谁先好谁先拿生产里更常用的是submit提交任务拿到Future,再用as_completed按「完成先后」而非「提交顺序」收结果。这样先跑完的先处理,慢的不拖累快的:fromconcurrent.futuresimportThreadPoolExecutor,as_completeddeffetch_status(url:str)-int:rrequests.get(url,timeout5)returnr.status_code urls[https://httpbin.org/delay/1,https://httpbin.org/status/404,https://httpbin.org/delay/2,https://not-a-real-host.invalid,# 故意让它失败]results,errors{},{}withThreadPoolExecutor(max_workers8)aspool:# future 反查 url,方便出错时知道是谁挂了future_to_url{pool.submit(fetch_status,u):uforuinurls}forfutureinas_completed(future_to_url):urlfuture_to_url[future]try:results[url]future.result()# 任务里抛的异常在这里重新抛出exceptExceptionase:errors[url]repr(e)print(成功:,results)print(失败:,errors)关键点在于future.result():如果任务函数内部抛了异常,异常被Future兜住,直到你调result()才重新抛出。所以用try/except包住它,就能把「成功的」和「失败的」分开收集,一个失败不影响其余任务。as_completed则保证谁先跑完谁先进循环,长尾任务不会阻塞短任务的处理。给任务加超时,别让一个卡死拖垮整批result(timeout...)可以给单个任务设等待上限。注意超时不会真的杀死线程(Python 线程无法强制中断),它只是让你别再干等:fromconcurrent.futuresimportTimeoutErrorasFutureTimeoutwithThreadPoolExecutor(max_workers8)aspool:futures{pool.submit(fetch_status,u):uforuinurls}forfutureinas_completed(futures,timeout10):# 整批 10 秒内必须收完urlfutures[future]try:print(url,future.result(timeout3))# 单个任务最多等 3 秒exceptFutureTimeout:print(url,该任务超时,先跳过)exceptExceptionase:print(url,失败:,e)真正的超时控制还得靠底层库自己支持,比如requests.get(url, timeout5)——线程池的 timeout 只是让主线程别一直傻等。线程池 vs 进程池:选错等于白干concurrent.futures还提供ProcessPoolExecutor,接口几乎一样,但适用场景相反:fromconcurrent.futuresimportProcessPoolExecutordefcpu_heavy(n:int)-int:returnsum(i*iforiinrange(n))# 纯计算,吃 CPU# CPU 密集用进程池,绕开 GIL,真正并行withProcessPoolExecutor(max_workers4)aspool:print(list(pool.map(cpu_heavy,[10**6]*4)))判断标准很简单:IO 密集(网络请求、读写文件、查数据库):线程主要在「等」,GIL 在等待时会释放,用ThreadPoolExecutor,max_workers可以开得比核心数大很多(几十到上百)。CPU 密集(加密、图像处理、大量数学计算):线程被 GIL 卡住无法并行,必须用ProcessPoolExecutor,max_workers一般设成 CPU 核心数。用线程池跑 CPU 密集任务,你会发现开多少线程都不快——因为 GIL 让它们始终只有一个在真正执行。小结独立的 IO 任务用ThreadPoolExecutor,一行with自动管理线程生命周期。优先submitas_completed,谁先完成谁先处理;用{future: 上下文}字典反查是哪个任务出的错。future.result()才是异常真正抛出的地方,用try/except包住它就能把成功与失败分流,单个失败不拖垮整批。result(timeout)只是让主线程别干等,真超时靠底层库(如requests的timeout)。一句话记忆:等 IO 用线程池,算 CPU 用进程池,收结果认准as_completed。