4. 线程池与异步任务：concurrent.futures 模块

说到并发编程，很多人第一反应就是自己手动创建线程、管理生命周期、处理异常。说实话，我早年也这么干过，结果代码写得又长又容易出 bug。后来接触了 concurrent.futures 模块，才发现原来线程池可以这么优雅。

这一章，我们就来聊聊 Python 里最实用的并发工具——ThreadPoolExecutor 和 Future 对象。我会结合自己踩过的坑，帮你把这块彻底搞明白。

4.1 为什么需要线程池？

先问个问题：你每次处理一个任务，都手动 threading.Thread(target=func).start()，然后自己管理 join、异常、结果收集？

这样做有几个痛点：

• 线程创建销毁开销大——每次 new 一个线程，操作系统都要分配栈空间、注册调度，频繁创建会拖垮性能。
• 并发数不可控——任务一多，线程数暴涨，CPU 上下文切换成本飙升，甚至可能把系统搞挂。
• 结果收集麻烦——你得用 Queue、Event 之类的同步原语手动传数据，代码可读性很差。

线程池就是来解决这些问题的。它预先创建一批线程，任务来了直接扔进去执行，线程复用，并发数可控，结果还能优雅地拿回来。

4.2 ThreadPoolExecutor 基础用法

ThreadPoolExecutor 是 concurrent.futures 模块的核心类。我习惯把它当作「线程版的 map/reduce」来用。

4.2.1 基本示例

from concurrent.futures import ThreadPoolExecutor
import time

def fetch_url(url):
    print(f"开始请求: {url}")
    time.sleep(2)  # 模拟网络IO
    return f"结果: {url}"

# 创建线程池，最大并发3
with ThreadPoolExecutor(max_workers=3) as executor:
    urls = ["http://a.com", "http://b.com", "http://c.com", "http://d.com"]
    results = executor.map(fetch_url, urls)

    for result in results:
        print(result)

你看，用 with 语句管理线程池生命周期，executor.map 自动把每个 URL 分配给空闲线程。代码简洁，逻辑清晰。

4.2.2 submit 方法与 Future 对象

map 适合批量提交同类型任务。但如果你需要更细粒度的控制——比如取消任务、设置超时、添加回调——那就得用 submit 了。

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(fetch_url, "http://example.com")
    # future 是一个 Future 对象
    print(future.done())       # 任务是否完成？
    result = future.result()   # 阻塞等待结果
    print(result)

Future 对象代表一个异步操作的结果。你可以把它理解成一个「未来的结果占位符」——任务还没执行完，但你可以先拿着这个对象，等需要结果时再取。

方法	说明	我踩过的坑
result(timeout=None)	阻塞等待结果，可设超时	不设 timeout 可能永久阻塞
done()	非阻塞，返回是否完成	别用它做精确判断，有竞态
cancel()	尝试取消任务	只能取消未开始的任务
add_done_callback(fn)	任务完成时自动回调	回调在哪个线程执行？注意线程安全
exception(timeout=None)	获取任务抛出的异常	不调用这个，异常会被吞掉

4.3 Future 对象与回调机制

回调是 Future 最强大的特性之一。你想想看，如果每个任务完成后都要手动调用 result() 去拿结果，那和同步编程有什么区别？回调让你可以「任务完成时自动触发后续逻辑」。

4.3.1 add_done_callback 实战

def on_complete(future):
    try:
        result = future.result()
        print(f"回调收到结果: {result}")
    except Exception as e:
        print(f"任务失败: {e}")

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(fetch_url, "http://slow-api.com")
    future.add_done_callback(on_complete)
    print("主线程继续做其他事...")
    # 这里不会阻塞

回调函数会在任务完成时，由线程池中的某个线程执行。注意：回调是在哪个线程执行的？ 答案是——完成任务的线程，或者主线程（如果任务已经完成）。所以回调里如果有共享数据，记得加锁。

4.3.2 多个 Future 的编排

实际项目中，经常需要等一批任务都完成后再做下一步。这时候可以用 concurrent.futures.wait 或 as_completed。

from concurrent.futures import wait, as_completed, FIRST_COMPLETED

futures = [executor.submit(fetch_url, url) for url in urls]

# 方式1：等所有完成
done, not_done = wait(futures, return_when=FIRST_COMPLETED)
print(f"已完成 {len(done)} 个")

# 方式2：谁先完成处理谁
for future in as_completed(futures):
    print(future.result())

as_completed 是我个人非常喜欢的方式。它返回一个迭代器，哪个任务先完成就先 yield 出来。这样你可以尽早处理结果，不用等最慢的那个。

4.4 线程池的核心流程

说了这么多，我们来画一张图，把整个流程串起来。

4.5 避坑指南与最佳实践

这部分是我这些年积累的血泪教训，你仔细看看，能省不少调试时间。

4.5.1 线程安全

线程池里的回调函数、任务函数，都可能被多个线程同时执行。如果你在回调里修改了共享变量，记得加锁。

from threading import Lock

counter = 0
lock = Lock()

def safe_callback(future):
    global counter
    with lock:
        counter += 1
        print(f"已完成 {counter} 个任务")

4.5.2 异常处理

任务函数里抛异常，默认不会打印到控制台。异常会被封装到 Future 里，等你调用 result() 时才抛出。

def risky_task(x):
    if x == 0:
        raise ValueError("x 不能为 0")
    return 10 / x

with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(risky_task, i) for i in [1, 0, 2]]
    for f in as_completed(futures):
        try:
            print(f.result())
        except Exception as e:
            print(f"任务异常: {e}")

4.5.3 避免死锁

线程池里的任务如果又往同一个线程池提交任务，就可能死锁。因为所有线程都在等新任务完成，但新任务排不进去。

4.5.4 合理设置 max_workers

场景	建议值	原因
纯 IO 密集型（网络请求、文件读写）	CPU 核心数 × 2~4	线程大部分时间在等待 IO，多开线程提高吞吐
混合型（少量计算 + 大量 IO）	CPU 核心数 × 2	平衡计算和等待
CPU 密集型	CPU 核心数	再多也没用，反而增加上下文切换
不确定场景	先压测，再调优	没有银弹，实测最靠谱

4.6 实战：并发下载器

最后，我们写一个完整的例子，把今天学的都用上。

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock
import time

class ConcurrentDownloader:
    def __init__(self, max_workers=5):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.results = []
        self.lock = Lock()
    
    def download(self, url):
        """下载单个文件"""
        try:
            resp = requests.get(url, timeout=10)
            resp.raise_for_status()
            return (url, len(resp.content), None)
        except Exception as e:
            return (url, 0, str(e))
    
    def on_complete(self, future):
        """回调：记录结果"""
        url, size, error = future.result()
        with self.lock:
            self.results.append((url, size, error))
            status = "成功" if error is None else f"失败: {error}"
            print(f"[{status}] {url} - {size} bytes")
    
    def run(self, urls):
        """批量下载"""
        futures = []
        for url in urls:
            future = self.executor.submit(self.download, url)
            future.add_done_callback(self.on_complete)
            futures.append(future)
        
        # 等待所有完成
        for f in as_completed(futures):
            pass  # 回调已经处理了结果
        
        return self.results

# 使用示例
if __name__ == "__main__":
    urls = [
        "http://example.com/file1",
        "http://example.com/file2",
        # ... 更多URL
    ]
    downloader = ConcurrentDownloader(max_workers=10)
    results = downloader.run(urls)
    print(f"共下载 {len(results)} 个文件")

这个例子展示了线程池 + 回调 + 线程安全锁的完整配合。你可以在实际项目中直接套用这个模式。