用 concurrent.futures 给批量任务加并发与超时控制

日常开发中经常遇到“批量”场景：批量请求几十个接口、批量下载文件、批量检查一堆 URL 是否可达。如果用 for 循环串行执行，10 个任务每个耗时 2 秒就是 20 秒，效率很低。

Python 标准库 concurrent.futures 提供了一个非常干净的并发抽象：ThreadPoolExecutor（线程池）和 ProcessPoolExecutor（进程池）。今天用一个完整的批量任务例子，讲讲怎么用它实现并发、超时控制和异常隔离。

一、基本用法：把串行改成并发

假设有一个耗时的函数 check(url)，串行写法：

results = []
for url in urls:
    results.append(check(url))   # 一个个等

改成线程池只需要几行：

from concurrent.futures import ThreadPoolExecutor

def check(url):
    ...

urls = ["https://a.example", "https://b.example"]

with ThreadPoolExecutor(max_workers=8) as pool:
    futures = [pool.submit(check, url) for url in urls]
    results = [f.result() for f in futures]

pool.submit() 把任务丢进线程池，立即返回一个 Future 对象；future.result() 在结果就绪时取出，阻塞等待。with 语句会在退出时自动调用 shutdown(wait=True)，确保所有任务完成。

二、两种提交方式：map 与 submit

• executor.map(func, iterable)：用法最像内置 map，返回结果与输入顺序一致，代码最短。
• executor.submit(func, arg) 配合 as_completed：更灵活，能做到“谁先完成谁先处理”，也方便单独设置超时。

对于需要逐个处理结果、或者任务耗时差异很大的场景，推荐 submit + as_completed。

三、超时控制：别让一个慢任务拖死整体

批量任务最大的坑是“个别任务 hang 住”。比如 100 个 URL 里有一个永远不响应，整体就被拖死了。concurrent.futures 有两层超时手段。

1. Future 层面的超时

from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError

with ThreadPoolExecutor(max_workers=8) as pool:
    future_to_url = {pool.submit(check, url): url for url in urls}
    for future in as_completed(future_to_url, timeout=30):
        url = future_to_url[future]
        try:
            status = future.result(timeout=5)   # 单个任务最多等 5 秒
            print(url, "->", status)
        except TimeoutError:
            print(url, "超时，已跳过")

注意区分两个 timeout：

• as_completed(..., timeout=30)：整个迭代最多等 30 秒，超时后抛出 TimeoutError，剩下的 Future 就不管了；
• future.result(timeout=5)：单个 Future 最多等 5 秒。

2. 任务函数内部的真实超时

线程池的超时有个细节：future.result(timeout=...) 超时只是“不等了”，线程本身不会被杀死，仍在后台跑。真正的超时控制要放在任务函数内部，比如给 requests.get() 传 timeout 参数：

import requests

def check(url):
    try:
        resp = requests.get(url, timeout=8)   # 连接加读取共 8 秒
        return resp.status_code
    except requests.RequestException as e:
        return f"失败: {e}"

两层配合：函数内部用库自带的超时做“真超时”，Future 层面用 result(timeout=...) 做“兜底”。

四、异常隔离：一个任务报错不影响其他

线程池里某个任务抛异常，不会崩掉主程序，异常会被 Future“吃掉”，等你调用 future.result() 时才重新抛出。所以处理模式很固定：

for future in as_completed(future_to_url):
    try:
        data = future.result()
    except Exception as e:
        print("任务失败:", e)
        continue
    # 正常处理 data

每个任务独立 try/except，一个挂了不影响其他结果的收集。

五、完整示例：批量检查 URL 状态

把上面的点串起来：

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError

URLS = [
    "https://a.example",
    "https://b.example",
    # ... 更多地址
]

def check(url):
    try:
        r = requests.get(url, timeout=8)
        return url, r.status_code
    except Exception as e:
        return url, f"异常: {type(e).__name__}"

def batch_check(urls, workers=8, per_task_timeout=10):
    ok, failed = [], []
    with ThreadPoolExecutor(max_workers=workers) as pool:
        future_map = {pool.submit(check, u): u for u in urls}
        try:
            for fut in as_completed(future_map, timeout=60):
                url = future_map[fut]
                try:
                    u, code = fut.result(timeout=per_task_timeout)
                except TimeoutError:
                    failed.append((url, "等待结果超时"))
                    continue
                except Exception as e:
                    failed.append((url, f"任务异常: {e}"))
                    continue
                (ok if code == 200 else failed).append((u, code))
        except TimeoutError:
            print("整体超时，剩余任务不再等待")
    return ok, failed

if __name__ == "__main__":
    ok, failed = batch_check(URLS)
    print(f"成功 {len(ok)} 个，失败 {len(failed)} 个")
    for u, code in failed:
        print("失败:", u, code)

这个模板里同时用到了：线程池并发、整体 60 秒超时、单个任务结果等待 10 秒兜底、函数内部 requests 的 8 秒真超时，以及每个任务独立的异常捕获。

六、几个实践建议

1. 线程数别乱设：IO 密集型任务（网络请求、文件读写）可以设大一点，比如 8 到 32；CPU 密集型任务线程数超过 CPU 核心数意义不大，考虑 ProcessPoolExecutor。
2. 永远给 IO 操作加超时：requests、socket、数据库连接都要设 timeout，不要只依赖 Future 超时兜底。
3. 注意共享资源的线程安全：多个线程往同一个 list 里 append 是安全的，但更复杂的数据结构要自己加锁。
4. 大任务量时分批提交：几万个任务一次性 submit 会占用大量内存，可以每 500 个一批提交处理。

小结

concurrent.futures 的核心就三招：submit 提交任务、as_completed 按完成顺序收结果、future.result(timeout=...) 做超时兜底；再配上任务函数内部的真实超时和异常隔离，就能写出健壮的批量并发代码。下次遇到“批量请求几十个接口”的场景，不妨试试这个模板。