大数据

ThreadPoolExecutor 知识点回顾

Posted by

demo01

import time
from concurrent.futures import ThreadPoolExecutor, as_completed


# 模拟一个I/O任务
def download_file(file_id):
    time.sleep(1)  # 假装在下载
    return f"文件{file_id}下载完成"

# 创建线程池（最多3个线程）
# 等待所有任务完成再退出
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务，立即返回Future对象
    future1 = executor.submit(download_file, 1)
    future2 = executor.submit(download_file, 2)

    # 当需要结果时调用.result() 会阻塞直到任务完成
    print(future1.result())
    print(future2.result())

demo02

# 通常我们有一堆任务，谁先完成就先处理谁，不按提交顺序
import time
from concurrent.futures import ThreadPoolExecutor, as_completed


def fetch_url(url):
    time.sleep({'a': 2, 'b': 1, 'c': 3}.get(url))
    return f"{url} 完成"


urls = ['a', 'b', 'c']

with ThreadPoolExecutor(max_workers=3) as executor:
    # 批量提交 用字典把 Future 和它的参数关联起来 以便知道结果来自哪个任务
    futures = {executor.submit(fetch_url, url): url for url in urls}

    # as_completed 迭代器：谁先完成就先yield谁
    for future in as_completed(futures):
        url = futures[future]  # 同过先完成发future获取对应的url
        try:
            result = future.result()
            print(f"[{url}] -> {result}")
        except Exception as e:
            print(f"[{url}] 出错：{e}")

demo03

import time
from concurrent.futures import ThreadPoolExecutor


def double(x):
    time.sleep(0.5)
    return x * 2


with ThreadPoolExecutor(max_workers=3) as executor:
    # map 接收可迭代对象，返回结果的生成器（保持输入顺序）
    results = executor.map(double, [1, 2, 3, 4, 5])
    # 输出按 1 到 5 的顺序打印，但任务是并发执行的。
    for num, res in zip([1, 2, 3, 4, 5], results):
        print(f"double({num}) = {res}")

demo04

from concurrent.futures import ThreadPoolExecutor, as_completed


def risky_task(n):
    if n == 2:
        raise ValueError("n不能为2")
    return n * 10


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

    for future in as_completed(futures):
        try:
            print(future.result())
        except Exception as e:
            print(f"任务失败：{e}")

demo05

# 导入线程池执行器，用于方便地创建和管理线程
from concurrent.futures import ThreadPoolExecutor
# 导入互斥锁，用于保护共享资源，防止多个线程同时修改
from threading import Lock

# 全局计数器，所有线程都会修改它
counter = 0
# 创建一个互斥锁对象
lock = Lock()


def increment():
    """每个线程要执行的任务：将 counter 累加 10 万次"""
    # 声明 counter 是全局变量，而不是局部变量
    global counter
    # 循环 100,000 次，每次对 counter 加 1
    for _ in range(100000): # 10w
        # 使用 with 语句获取锁，等效于 lock.acquire() 和 lock.release()
        # 任何时刻只有一个线程能进入这个代码块，其他线程会在此等待
        with lock:  # 不加锁，最终结果会小于预期
            counter += 1  # 临界区：该操作不是原子的，需要加锁保护


# 使用 with 创建线程池，max_workers=5 表示池中最多同时运行 5 个线程
with ThreadPoolExecutor(max_workers=5) as executor:
    # 提交 10 个同样的任务到线程池
    for _ in range(10): # 10 * 10w = 100w
        # submit 方法将函数提交给线程池执行，立即返回 Future 对象（这里没有保存）
        executor.submit(increment)

# 所有任务完成后（因为 with 会等待线程池关闭），打印最终计数
print(counter)  # 输出：1000000

demo06

import time
from concurrent.futures import ThreadPoolExecutor


def callback(future):
    # 无论成功或失败这里都能拿到结果或异常
    try:
        result = future.result()
        print(f"回调收到：{result}")
    except Exception as e:
        print(f"回调捕捉到异常：{e}")


with ThreadPoolExecutor(max_workers=2) as executor:
    f = executor.submit(lambda : 1/0) # 会抛异常
    f.add_done_callback(callback)
    # 主线可以继续做其他实现
    time.sleep(0.5)

demo07

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed


def fetch_url(url):
    try:
        resp = requests.get(url, timeout=5)
        return url, resp.status_code, len(resp.content)
    except Exception as e:
        return url, None, str(e)


urls = ["https://httpbin.org/delay/1"] * 5  # 5个慢请求

with ThreadPoolExecutor(max_workers=10) as executor:
    future_to_url = {executor.submit(fetch_url, u): u for u in urls}

    for future in as_completed(future_to_url):
        url, status, length = future.result()
        print(f"{url} -> 状态：{status}，长度：{length}")

ThreadPoolExecutor vs ProcessPoolExecutor

特性ThreadPoolExecutorProcessPoolExecutor
适用任务I/O 密集型CPU 密集型
GIL 影响受限于 GIL，CPU 并行不行绕过 GIL，真正多核
内存/开销轻量，共享内存进程隔离，开销大
数据共享需锁/队列需 multiprocessing 机制

选择口诀：
“读写网络和磁盘，线程池里跑得欢；复杂计算要并行，进程池子才可行。”

常见坑与最佳实践

  1. max_workers 不是越大越好
    • I/O 密集：可设 10~50，甚至更多，但受限于系统和目标服务器。
    • 可以用 (cpu_count() * 5) 作为初始参考。
  2. 不要在任务中修改共享状态不加锁
    • 如全局列表、字典，多线程同时操作容易出 bug。
  3. Future.result() 阻塞主线程
    • 需考虑是否需要先完成所有任务才能继续。需要实时响应就用 as_completed + 回调。
  4. 用 with 语句保证资源释放
    • 即使发生异常，线程池也会等待任务完成并关闭。
  5. 任务内异常必须捕获
    • 否则只能通过 future.result() 才抛出，容易被忽略。
下一篇
dataclass于pydantic回顾
上一篇 shell 知识点回顾