在Python并发编程里，我们经常遇到这样的需求：同时向多个服务发起相同的查询，或者启动若干个彼此独立的任务，只要其中任意一个最先完成并返回有效结果，其余任务就可以不再关心。这种模式通常被称为「竞速」或「先到先得」。标准库concurrent.futures模块为此提供了非常直接的工具，核心在于as_completed函数和Future对象的取消机制。
为什么需要优先获取最快结果
假设你在做一个聚合搜索功能，后端同时请求三个不同的搜索引擎API。用户只想要最快出现的那条结果，如果顺序等待第一个API超时再试第二个，体验会非常差。使用线程池或进程池并发发出请求后，如果采用Executor.map或者依次调用future.result()，程序会按照任务提交顺序阻塞，也就是必须等第一个任务结束才能看第二个，这完全丧失了并发的意义。
另一种做法是等所有任务都完成再挑最快的，但这会浪费大量时间在慢任务上，并且已经拿到可用结果后还要空耗资源。理想方案是：谁先完成且结果合法，就立刻采用，并尽量中断其他还在跑的任务。这不仅能缩短响应时间，也能降低CPU、网络带宽和连接数的开销。
concurrent.futures的核心工具
concurrent.futures里有两个关键概念：Executor（执行器，如ThreadPoolExecutor、ProcessPoolExecutor）和Future（未来对象，代表尚未完成的异步结果）。提交任务后会得到一个Future，它包含done、result、cancel等方法。as_completed是一个接收future列表并返回迭代器的函数，迭代器会在某个future完成时立刻产出该future，顺序完全由实际完成时间决定。
下面的代码展示了基本用法：我们向线程池提交三个模拟耗时不同的任务，用as_completed遍历，一旦拿到第一个成功结果就记录并取消其余future。注意cancel并不保证一定生效，只有任务还处于未开始或运行中但未真正执行到不可中断点时才有可能；对于线程中的纯Python阻塞调用，cancel通常无效，因此更常见的做法是忽略其余结果，或在任务内部定期检查退出标志。
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random
def fake_task(name):
# 模拟不同耗时的网络请求
sleep_time = random.uniform(0.5, 3.0)
time.sleep(sleep_time)
return f"结果来自{name}, 耗时{sleep_time:.2f}秒"
def get_fastest_result():
tasks = {}
with ThreadPoolExecutor(max_workers=3) as executor:
for i in range(3):
future = executor.submit(fake_task, f"服务{i+1}")
tasks[future] = i
for future in as_completed(tasks):
try:
result = future.result()
except Exception as e:
print("任务出错:", e)
continue
print("最快结果:", result)
# 拿到第一个有效结果后，尝试取消其余任务
for f in tasks:
if f is not future:
f.cancel()
break
if __name__ == "__main__":
get_fastest_result()
取消任务的局限与改进
上面代码中的cancel对线程池往往只是「标记取消」，如果任务已经在执行且处于time.sleep中，它并不会真正中断睡眠。因此在真实项目中，如果任务是IO密集型且使用线程，我们通常依靠「忽略结果」来节约后续处理成本，而不是强求终止底层调用。如果是CPU密集型且使用ProcessPoolExecutor，cancel在任务尚未被工作进程取走时可以有效避免启动新进程。
更健壮的做法是把任务函数设计成可协作取消的。例如在线程中周期性检查一个共享的threading.Event，或者在网络请求时使用带超时的调用，并在拿到首结果后通过事件通知其他任务提前返回。下面给出一个带退出事件的示例，展示如何让慢任务主动让路。
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
stop_event = threading.Event()
def cooper_task(name):
for i in range(10):
if stop_event.is_set():
return f"{name}被提前终止"
time.sleep(0.3)
return f"{name}正常完成"
with ThreadPoolExecutor(max_workers=3) as ex:
futures = [ex.submit(cooper_task, f"任务{n}") for n in range(3)]
for f in as_completed(futures):
print("得到:", f.result())
stop_event.set()
break
进程池与线程池的选择
如果任务是计算密集型，应优先使用ProcessPoolExecutor，这样能绕过GIL并利用多核。此时as_completed同样适用，且cancel在任务排队阶段更有效。如果是网络IO密集型，ThreadPoolExecutor足够，因为大多数时间花在等待socket上，线程切换成本很低。无论哪种池，都建议通过with语句管理生命周期，避免资源泄漏。
另外，Python 3.8之后concurrent.futures增加了取消时的回调支持，可以更精细地处理清理逻辑。但在获取最快结果这个场景下，核心思路不变：并发提交、按完成顺序消费、首结果到达即终止后续关注。下表对比了三种常见策略：
| 策略 | 响应速度 | 资源利用 | 实现复杂度 |
|---|---|---|---|
| 顺序执行 | 最慢 | 低 | 最简单 |
| 全量等待再筛选 | 中等 | 浪费 | 中等 |
| as_completed竞速 | 最快 | 最优 | 稍高 |
实际场景中的注意事项
在真实系统里，最快结果未必是最优结果。比如多数据源返回的格式不同，需要先校验结构再采纳。可以在future.result()之后加一层验证，若不符合预期则继续从as_completed取下一个，而不是立刻break。这样既能保证速度，也能兼顾质量。
还要注意异常隔离：某个任务抛异常不应影响其他任务，as_completed依然会产出该future，调用result时会重新抛出，因此必须用try捕获。最后，若并发量很大，应限制Executor的max_workers，防止线程或进程爆炸式增长拖垮宿主机。
Python并发编程concurrent_futures修改时间：2026-08-03 20:21:38