python线程池

ThreadPoolExecutor 是 Python 标准库 concurrent.futures 提供的线程池执行器。它会预先管理一组线程，让你把多个任务提交给线程池并发执行，而不用手动创建、启动和回收线程。

线程池可以理解成：提前准备一组线程，任务来了就交给空闲线程处理，用完线程不销毁，继续等待下一个任务。

1. 它解决的问题

不用线程池的话，一个任务要并发，就得自己起一条线程：

import threading

t = threading.Thread(target=work, args=("A",))
t.start()
t.join()

任务少的时候还行，任务一多就有三个问题：

• 创建和销毁有开销。 线程不是免费的，每个都要一条独立的栈。一百个任务就是一百次创建 + 一百次销毁。
• 数量不可控。 谁都能 Thread 一下，一千个请求就起一千条线程，机器先垮了。
• 拿不到返回值。 threading.Thread 没有返回值，结果得自己往队列或者全局变量里塞。

线程池把这三件事一次解决：线程复用、数量封顶、结果装在 Future 里返回。

2. 最小例子

from concurrent.futures import ThreadPoolExecutor

def work(x):
    return x * 2

with ThreadPoolExecutor(max_workers=2) as pool:
    pool.submit(work, 1)
    pool.submit(work, 2)
    pool.submit(work, 3)

# 离开 with 后，线程池执行 shutdown()
# 等待任务完成，然后线程退出

这里最值得说的是那个 with。ThreadPoolExecutor 支持上下文管理器，退出代码块时会自动调用 shutdown()——先把提交的任务等完，再让线程退出。不写 with 就得自己记着 pool.shutdown()，忘了就是线程悬在那儿。

3. 提交任务：submit 和 map

submit：一次提交一个，返回 Future

from concurrent.futures import ThreadPoolExecutor

def add(a, b):
    return a + b

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(add, 1, 2)
    result = future.result()

print(result)  # 3

submit 的签名是 submit(fn, *args, **kwargs)——函数名后面直接跟参数，位置参数、关键字参数都这么传。

map：批量提交

from concurrent.futures import ThreadPoolExecutor
import time

def work(name):
    print(f"开始处理 {name}")
    time.sleep(2)
    print(f"完成 {name}")

tasks = ["A", "B", "C", "D", "E"]

with ThreadPoolExecutor(max_workers=2) as pool:
    pool.map(work, tasks)

pool.map(work, tasks) 的意思是：把 tasks 里的每个元素依次传给 work，再由线程池并发执行。

这段代码里 5 个任务、2 条线程、每个任务 2 秒，所以是 3 批 × 2 秒 ≈ 6 秒，不是 10 秒。map 的接口长得跟内置 map() 一模一样，但内置 map 是惰性串行，这里是全部提交进去并发跑。

from concurrent.futures import ThreadPoolExecutor

def download(url):
    print(f"正在处理：{url}")
    return f"{url} 完成"

urls = ["a.com", "b.com", "c.com"]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(download, urls)

for result in results:
    print(result)

两个的差别

	submit	map
一次提交几个	一个	一批
返回什么	Future 对象	结果迭代器
结果顺序	谁先完成谁先有	按提交顺序
参数	每个任务的参数可以完全不同	每个任务只有同一个函数、一组对应参数
什么时候用	任务异构、要单独处理某个结果	同一件事跑一批数据

map 还有个容易卡住的地方：它是按提交顺序产出结果的。

results = executor.map(work, tasks)   # tasks[0] 要 10 秒，其余都是 1 秒
for r in results:
    print(r)

tasks[0] 没完成之前，一行都不会打印——哪怕后面四个早就跑完了，结果也只能排在它后面等着。想按完成顺序拿，得用 as_completed（下一节）。

4. Future：可能尚未完成的任务

submit 返回的 Future 代表一个可能尚未完成的任务。它这一组方法要分清楚：

future.result()       # 获取结果，必要时等待
future.done()         # 是否完成
future.cancel()       # 尝试取消
future.exception()    # 获取异常

result() 会阻塞

这是 Future 最核心的一条：result() 在任务没完成时会一直等。 所以拿到 Future 之后马上 result()，等于把并发又变回了串行：

futures = [executor.submit(download, u) for u in urls]

for f in futures:
    print(f.result())     # ❌ 按顺序一个个等，并发白做了

正确的做法是先全部提交，再统一收结果。

异常藏在 Future 里，不 result() 就没人知道

def boom():
    raise ValueError("炸了")

with ThreadPoolExecutor() as pool:
    future = pool.submit(boom)

    future.done()         # 可能 True 也可能 False，任务跑得快就完了
    future.exception()    # ValueError("炸了") —— 注意它是**返回**异常，不是抛出
    future.result()       # 这里才真的抛出来

exception() 和 result() 的行为不一样：前者把异常当返回值给你，后者把它抛出来。

这条要记住：submit 提交的任务如果抛异常，异常会被存在 Future 里，只有你调 result()（或 exception()）才会知道。 从头到尾不看一眼，这个错误就是静默丢失的——日志里什么都没有。map 也一样，异常会在你迭代到那个元素时才抛出来。

cancel() 是”尝试取消”

它只能取消还没开始执行的任务：

• 任务还在队列里排着 → 取消成功，cancel() 返回 True
• 任务已经被某条线程捞起来在跑了 → 取消不了，返回 False

已经在跑的线程没法从外面喊停，这一点和协程不一样——协程可以在 await 处被 cancel() 打断（见 asyncio 那篇），线程只能等它自己跑完。

它不是 asyncio.Future

concurrent.futures.Future 和 asyncio.Future 是两个不同的类，名字一样但不能互相 await。前者是给线程池用的，后者是给事件循环用的。混用会得到 TypeError。

5. 怎么收结果：map 按顺序，as_completed 按完成顺序

from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(download, url): url for url in urls}

    for future in as_completed(futures):
        url = futures[future]
        try:
            print(url, future.result())
        except Exception as e:
            print(url, "失败:", e)

as_completed(futures) 是一个迭代器，谁先完成谁先出来，跟提交顺序无关。

用字典把 Future 映射回原始参数是个标准写法——因为 as_completed 只给你 Future，不给你当初传进去的 url。用列表按下标找也行，但字典更直接。

另外注意 try 包在 future.result() 外面：这是唯一能把单个任务的失败和整批任务分开处理的地方。

6. max_workers 设多少

max_workers 是线程池的上限，也就是同时最多有几条线程在跑。

默认值：Python 3.8 起是 min(32, (os.cpu_count() or 1) + 4)。 3.8 之前是 cpu_count() * 5，官方觉得在很多核的机器上大得不合理，改掉了。这个默认值的设计意图写得很清楚：

• cpu_count() + 4 保证至少 5 条，够 I/O 密集型任务用
• min(32, ...) 封顶，避免 128 核的机器上默默开出 640 条线程

怎么设：

任务类型	怎么设
I/O 密集（网络请求、数据库、读写文件）	可以设大一些，几十到几百都行。线程大部分时间在等，不占 CPU
CPU 密集	设大基本没用。 受 GIL 限制，纯 Python 计算同一时刻只有一个线程在跑

CPU 密集的任务应该换 ProcessPoolExecutor，不是把 max_workers 调大。这条和 asyncio 那篇里线程 / 进程 / 协程的选型表是同一个结论。

(一个例外：如果耗时的那段计算发生在释放 GIL 的 C 扩展里——NumPy 的大数组运算之类——线程是能真正并行的，文档里”最多用 32 个核”说的就是这种。)

7. shutdown：with 退出时到底做了什么

pool.shutdown(wait=True)

shutdown(wait=True) 表示：

1. 不再接受新任务。
2. 等待已经提交的任务完成。
3. 让工作线程退出。
4. 回收线程池资源。

第 4 步对应开头那句”用完线程不销毁”——这里的”用完”指的是用完一次任务，不是”整个线程池都不用了”。 一条线程处理完一个任务后不立刻销毁，回到池子里等下一个任务；只有 shutdown() 之后才真的退出。

wait=False 就是”不等了，先返回”——已提交的任务还会跑完，但你不等它们。

with 块退出时调用的就是 shutdown(wait=True)，和上面四步完全一样。

8. 它和 asyncio 的关系：to_thread 背后就是线程池

上一篇里用 asyncio.to_thread() 把同步调用扔出去，那个”扔到线程里”的线程，就来自一个线程池。

to_thread 的实现基本就是一层包装：

loop = asyncio.get_running_loop()
ctx = contextvars.copy_context()
func_call = functools.partial(ctx.run, func, *args, **kwargs)
return await loop.run_in_executor(None, func_call)

关键在 run_in_executor(None, ...) 的那个 None：它表示”用事件循环的默认执行器”，而那个默认执行器就是线程池，默认线程数同样是 min(32, (os.cpu_count() or 1) + 4)。

这件事有个实际后果：在一台 2 核的机器上，这个池子只有 6 条线程。 如果你的异步程序里有大量 to_thread 调用在同时等，它们会在池子前面排队，看起来像”异步没生效”。

需要更大的池子，就把默认执行器换掉：

import asyncio
from concurrent.futures import ThreadPoolExecutor

loop = asyncio.get_running_loop()
loop.set_default_executor(ThreadPoolExecutor(max_workers=64))

所以两者的关系是：

	直接用它	通过 asyncio 用
入口	ThreadPoolExecutor	asyncio.to_thread(...) / run_in_executor(...)
谁来调度	你自己提交、自己收结果	事件循环，await 一下就行
池子	自己建的	事件循环的默认池（也可以自己建了传进去）
适合	纯同步代码里做并发	已经在一个 async def 里

在 async def 里，优先用 asyncio.to_thread，别自己建池子再 run_in_executor——除非你需要控制池子大小，那时候才手动建一个传进去。

9. 小结

四件事：

一、线程池解决的是”线程复用 + 数量封顶 + 能拿返回值”。 with 退出自动 shutdown(wait=True)，等任务跑完再让线程退出。

二、submit 给你 Future，map 给你按提交顺序的结果迭代器。 要按完成顺序拿就用 as_completed。

三、异常藏在 Future 里，不调 result() 就静默丢失。 而且 result() 会阻塞，所以要先全提交、再统一收。

四、它是 I/O 密集的工具，不是 CPU 密集的。 CPU 密集换 ProcessPoolExecutor；在 async def 里就用 asyncio.to_thread，两者背后是同一件事。

参考

• concurrent.futures 总览：https://docs.python.org/zh-cn/3/library/concurrent.futures.html
• ThreadPoolExecutor：https://docs.python.org/zh-cn/3/library/concurrent.futures.html#threadpoolexecutor
• Future 对象的方法：https://docs.python.org/zh-cn/3/library/concurrent.futures.html#future-objects
• asyncio.to_thread：https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio.to_thread
• loop.set_default_executor：https://docs.python.org/zh-cn/3/library/asyncio-eventloop.html#asyncio.loop.set_default_executor