Python 线程池 ThreadPoolExecutor

ThreadPoolExecutor 自动管理线程资源，适合批量并发任务。

基本使用

Python

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    time.sleep(1)
    return n * 2

# 创建线程池
with ThreadPoolExecutor(max_workers=4) as executor:
    # 提交任务
    future = executor.submit(task, 5)
    result = future.result()
    print(result)  # 10

submit 提交任务

Python

from concurrent.futures import ThreadPoolExecutor

def process(name):
    return f"处理完成: {name}"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = []
    for name in ['A', 'B', 'C', 'D', 'E']:
        future = executor.submit(process, name)
        futures.append(future)

    # 获取所有结果
    for future in futures:
        print(future.result())

map 批量映射

Python

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n ** 2

numbers = [1, 2, 3, 4, 5]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(square, numbers)
    print(list(results))  # [1, 4, 9, 16, 25]

Future 对象

Python

from concurrent.futures import ThreadPoolExecutor
import time

def slow_task(n):
    time.sleep(n)
    return n

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(slow_task, 2)

    # 检查状态
    print(future.done())      # False（未完成）

    # 设置超时等待
    result = future.result(timeout=5)
    print(future.done())      # True
    print(result)             # 2

    # 获取异常
    # future.exception() 返回异常对象

回调函数

Python

from concurrent.futures import ThreadPoolExecutor

def task(n):
    return n * 2

def callback(future):
    print(f"回调结果: {future.result()}")

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(task, 5)
    future.add_done_callback(callback)
    # 任务完成后自动调用回调

as_completed 按完成顺序获取

Python

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def task(n):
    time.sleep(n)
    return n

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(task, i): i for i in [3, 1, 2]}

    # 按完成顺序获取结果
    for future in as_completed(futures):
        print(f"任务 {futures[future]} 完成，结果: {future.result()}")

# 输出顺序: 1, 2, 3（按完成时间）

异常处理

Python

from concurrent.futures import ThreadPoolExecutor

def risky_task(n):
    if n == 3:
        raise ValueError("数字3不被允许")
    return n * 2

with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(risky_task, i) for i in range(5)]

    for future in futures:
        try:
            result = future.result()
            print(f"成功: {result}")
        except Exception as e:
            print(f"异常: {e}")

shutdown 控制

Python

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=3)

# 提交任务
futures = [executor.submit(lambda x: x**2, i) for i in range(5)]

# 关闭线程池
executor.shutdown(wait=True)   # 等待所有任务完成
# executor.shutdown(wait=False) # 不等待，立即关闭
# executor.shutdown(cancel_futures=True) # 取消未执行任务

上下文管理

Python

from concurrent.futures import ThreadPoolExecutor

# 推荐使用 with 语句
with ThreadPoolExecutor(max_workers=4) as executor:
    # 自动管理资源，退出时自动 shutdown
    results = executor.map(lambda x: x**2, range(10))

print(list(results))

方法对比

方法	说明	返回值
submit(fn, *args)	提交单个任务	Future
map(fn, *iterables)	批量映射	迭代器
shutdown(wait)	关闭线程池	None

Future 方法

方法	说明
result(timeout)	获取结果
exception()	获取异常
done()	是否完成
add_done_callback(fn)	添加回调

要点总结

ThreadPoolExecutor(max_workers=N) 创建线程池
submit() 提交单个任务，返回 Future
map() 批量处理，返回结果迭代器
as_completed() 按完成顺序获取结果
使用 with 语句自动管理资源
Future 对象提供任务状态和结果查询
线程池适合 I/O 密集型任务的批量处理

📝 发现内容有误？点击此处直接编辑