concurrent.futures 实战指南

from concurrent.futures import ThreadPoolExecutor
import time

def fetch_data(url: str) -> dict:
    """模拟网络请求"""
    time.sleep(1)  # 模拟 IO 等待
    return {"url": url, "status": 200, "data": f"response from {url}"}

# 推荐：使用 with 语句自动管理生命周期
with ThreadPoolExecutor(max_workers=5) as executor:
    # 提交单个任务
    future = executor.submit(fetch_data, "https://api.example.com/users")
    # 获取结果（阻塞直到任务完成）
    result = future.result()
    print(result)  # {'url': 'https://api.example.com/users', 'status': 200, ...}

import os
from concurrent.futures import ThreadPoolExecutor

# 规则一：IO 密集型任务，max_workers 可以远大于 CPU 核心数
# 经验公式：max_workers = min(32, CPU核心数 * 5)  （Python 3.8+ 默认值）
io_workers = min(32, (os.cpu_count() or 1) + 4)
print(f"IO 密集型建议线程数: {io_workers}")

# 规则二：如果任务有明显的等待时间，可以更大
# 例如：每个请求耗时 2 秒，需要 100 个请求
# 1 个线程：200 秒
# 10 个线程：20 秒
# 50 个线程：4 秒（受网络带宽限制）
# 100 个线程：2 秒（理想情况）

# 规则三：注意系统限制（文件描述符、连接数等）

from concurrent.futures import ThreadPoolExecutor, as_completed

def process_url(url: str) -> dict:
    """模拟处理 URL"""
    import time
    time.sleep(0.5)
    return {"url": url, "length": len(url)}

urls = [
    "https://api.example.com/users",
    "https://api.example.com/posts",
    "https://api.example.com/comments",
    "https://api.example.com/albums",
    "https://api.example.com/photos",
]

with ThreadPoolExecutor(max_workers=3) as executor:
    # 方式一：submit + as_completed（按完成顺序获取结果）
    future_to_url = {
        executor.submit(process_url, url): url
        for url in urls
    }

    for future in as_completed(future_to_url):
        url = future_to_url[future]
        try:
            result = future.result()
            print(f"[完成] {url} -> {result}")
        except Exception as e:
            print(f"[失败] {url} -> {e}")

from concurrent.futures import ThreadPoolExecutor
import threading

def show_thread_info(task_name: str) -> str:
    """展示线程池中线程的复用"""
    thread_name = threading.current_thread().name
    return f"任务 {task_name} 在线程 {thread_name} 中执行"

# max_workers=2 表示池中只有 2 个工作线程
with ThreadPoolExecutor(max_workers=2) as executor:
    # 提交 5 个任务，但只有 2 个线程，任务会排队
    futures = [
        executor.submit(show_thread_info, f"Task-{i}")
        for i in range(5)
    ]

    for future in futures:
        print(future.result())

# 输出示例（注意线程名的复用）：
# 任务 Task-0 在线程 ThreadPoolExecutor-0_0 中执行
# 任务 Task-1 在线程 ThreadPoolExecutor-0_1 中执行
# 任务 Task-2 在线程 ThreadPoolExecutor-0_0 中执行  ← 复用
# 任务 Task-3 在线程 ThreadPoolExecutor-0_1 中执行  ← 复用
# 任务 Task-4 在线程 ThreadPoolExecutor-0_0 中执行  ← 复用

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n: int) -> bool:
    """判断一个数是否为素数（CPU 密集型）"""
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False
    for i in range(3, int(math.sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True

with ProcessPoolExecutor() as executor:
    # 默认 max_workers = CPU 核心数
    result = executor.submit(is_prime, 999999937).result()
    print(f"999999937 是素数: {result}")  # True

from concurrent.futures import ProcessPoolExecutor

# ---- 可以序列化的 ----
def top_level_function(x: int) -> int:
    """模块级函数：可以序列化"""
    return x * x

class PicklableClass:
    """可序列化的类"""
    def __call__(self, x: int) -> int:
        return x * x

# ---- 不能序列化的 ----
class MyTask:
    def process(self, x: int) -> int:
        """实例方法：不能直接序列化"""
        return x * x

    @staticmethod
    def static_process(x: int) -> int:
        """静态方法：可以序列化"""
        return x * x

# 正确用法
with ProcessPoolExecutor() as executor:
    # 模块级函数：OK
    executor.submit(top_level_function, 42)

    # 可调用对象实例：OK
    executor.submit(PicklableClass(), 42)

    # 静态方法：OK
    executor.submit(MyTask.static_process, 42)

    # 实例方法：会报错！
    # executor.submit(MyTask().process, 42)  # AttributeError

from concurrent.futures import ProcessPoolExecutor
import time
import numpy as np

def process_array(arr: np.ndarray) -> float:
    """处理 NumPy 数组"""
    return arr.mean()

# 小数组：序列化开销可忽略
small_arr = np.random.rand(100)

# 大数组：序列化开销显著
big_arr = np.random.rand(10_000_000)

with ProcessPoolExecutor() as executor:
    # 小数组：快
    start = time.perf_counter()
    executor.submit(process_array, small_arr).result()
    print(f"小数组耗时: {time.perf_counter() - start:.4f}s")

    # 大数组：序列化/反序列化可能比计算本身还慢
    start = time.perf_counter()
    executor.submit(process_array, big_arr).result()
    print(f"大数组耗时: {time.perf_counter() - start:.4f}s")

from concurrent.futures import ProcessPoolExecutor
import random

# 每个工作进程初始化时执行的函数
def init_worker():
    """在工作进程中初始化随机种子，避免多进程随机数重复"""
    random.seed()

def random_task(n: int) -> float:
    """生成随机数"""
    return random.random() * n

with ProcessPoolExecutor(
    max_workers=4,
    initializer=init_worker,    # 初始化函数
    initargs=(),                # 初始化参数（元组）
) as executor:
    results = list(executor.map(random_task, [10, 20, 30, 40]))
    print(results)

from concurrent.futures import ThreadPoolExecutor
import time

def slow_task(seconds: float) -> str:
    """模拟耗时任务"""
    time.sleep(seconds)
    return f"完成，耗时 {seconds} 秒"

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

    # 1. 查询状态
    print(f"是否已完成: {future.done()}")      # False
    print(f"是否已取消: {future.cancelled()}")  # False

    # 2. 非阻塞检查（设置超时）
    try:
        result = future.result(timeout=1.0)  # 1 秒内没完成则抛 TimeoutError
    except TimeoutError:
        print("任务尚未完成")

    # 3. 阻塞等待结果
    result = future.result()  # 阻塞直到完成
    print(f"结果: {result}")
    print(f"是否已完成: {future.done()}")  # True

from concurrent.futures import ThreadPoolExecutor

def success_task() -> str:
    return "成功"

def failing_task() -> str:
    raise ValueError("任务执行失败")

with ThreadPoolExecutor(max_workers=2) as executor:
    # 成功的任务
    f1 = executor.submit(success_task)
    print(f1.result())  # 输出: "成功"

    # 失败的任务
    f2 = executor.submit(failing_task)
    try:
        f2.result()  # 抛出 ValueError
    except ValueError as e:
        print(f"捕获到任务异常: {e}")  # 捕获到任务异常: 任务执行失败

    # 带超时的 result
    f3 = executor.submit(lambda: __import__('time').sleep(10))
    try:
        f3.result(timeout=0.5)
    except TimeoutError:
        print("等待超时")

from concurrent.futures import ThreadPoolExecutor
import time

def fetch_url(url: str) -> dict:
    """模拟网络请求"""
    time.sleep(1)
    if "error" in url:
        raise ConnectionError(f"连接失败: {url}")
    return {"url": url, "status": 200}

def on_complete(future):
    """任务完成回调——无论成功或失败都会被调用"""
    try:
        result = future.result()
        print(f"  [回调-成功] {result['url']} -> {result['status']}")
    except Exception as e:
        print(f"  [回调-失败] {e}")

urls = [
    "https://api.example.com/users",
    "https://api.example.com/error",
    "https://api.example.com/posts",
]

with ThreadPoolExecutor(max_workers=3) as executor:
    for url in urls:
        future = executor.submit(fetch_url, url)
        future.add_done_callback(on_complete)

    # 回调在提交任务的工作线程中执行，不要在回调中做耗时操作

# 输出：
#   [回调-成功] https://api.example.com/users -> 200
#   [回调-失败] 连接失败: https://api.example.com/error
#   [回调-成功] https://api.example.com/posts -> 200

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random

def variable_task(task_id: int) -> dict:
    """耗时随机的任务"""
    duration = random.uniform(0.5, 3.0)
    time.sleep(duration)
    return {"id": task_id, "duration": round(duration, 2)}

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {
        executor.submit(variable_task, i): i
        for i in range(10)
    }

    print("=== 按完成顺序 ===")
    for future in as_completed(futures):
        task_id = futures[future]
        result = future.result()
        print(f"  任务 {task_id} 完成，耗时 {result['duration']}s")

from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED
import time

def task(task_id: int, duration: float) -> dict:
    time.sleep(duration)
    if task_id == 3:
        raise RuntimeError(f"任务 {task_id} 失败")
    return {"id": task_id, "duration": duration}

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(task, i, i * 0.5) for i in range(1, 6)]

    # 等待所有任务完成（默认行为）
    done, not_done = wait(futures, return_when=ALL_COMPLETED)
    print(f"全部完成: {len(done)} 个任务")

    # 等待第一个完成
    # done, not_done = wait(futures, return_when=FIRST_COMPLETED)
    # print(f"首个完成，还有 {len(not_done)} 个未完成")

    # 等待第一个异常（如果没有异常，等待全部完成）
    # done, not_done = wait(futures, return_when=FIRST_EXCEPTION)

    # 设置超时
    # done, not_done = wait(futures, timeout=2.0, return_when=ALL_COMPLETED)
    # print(f"2 秒内完成: {len(done)} 个，未完成: {len(not_done)} 个")

from concurrent.futures import ThreadPoolExecutor

def task_a(x: int) -> str:
    return f"A: {x * 2}"

def task_b(x: int) -> str:
    return f"B: {x * 3}"

with ThreadPoolExecutor(max_workers=2) as executor:
    # submit 可以提交不同的函数
    f1 = executor.submit(task_a, 10)
    f2 = executor.submit(task_b, 20)
    f3 = executor.submit(task_a, 30)

    # 按需获取结果
    print(f1.result())  # A: 20
    print(f2.result())  # B: 60
    print(f3.result())  # A: 60

from concurrent.futures import ThreadPoolExecutor

def process(x: int) -> str:
    return f"处理: {x * 2}"

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

with ThreadPoolExecutor(max_workers=3) as executor:
    # map 对同一函数批量提交，结果按提交顺序返回
    results = list(executor.map(process, data))
    print(results)
    # ['处理: 2', '处理: 4', '处理: 6', '处理: 8', '处理: 10']

    # map 支持多个可迭代对象（类似内置 map）
    def add(a: int, b: int) -> int:
        return a + b

    sums = list(executor.map(add, [1, 2, 3], [10, 20, 30]))
    print(sums)  # [11, 22, 33]

    # map 支持 timeout
    # results = list(executor.map(process, data, timeout=5.0))

from concurrent.futures import ThreadPoolExecutor

def risky_task(x: int) -> int:
    if x == 3:
        raise ValueError(f"无效值: {x}")
    return x * 10

with ThreadPoolExecutor(max_workers=3) as executor:
    # map 的异常陷阱：异常不会在提交时抛出
    # 而是在迭代到该位置时才抛出
    results = executor.map(risky_task, [1, 2, 3, 4, 5])

    # 迭代时，到 x=3 才会抛异常，x=4 和 x=5 的结果丢失
    try:
        for r in results:
            print(r)
    except ValueError as e:
        print(f"异常: {e}")

    # 输出：
    # 10
    # 20
    # 异常: 无效值: 3
    # 4 和 5 的结果丢失！

# 更安全的做法：用 submit + as_completed
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(risky_task, x): x for x in [1, 2, 3, 4, 5]}

    for future in as_completed(futures):
        x = futures[future]
        try:
            result = future.result()
            print(f"  x={x} -> {result}")
        except ValueError as e:
            print(f"  x={x} 失败: {e}")

    # 输出（所有结果都能获取）：
    #   x=1 -> 10
    #   x=2 -> 20
    #   x=3 失败: 无效值: 3
    #   x=4 -> 40
    #   x=5 -> 50

"""批量 URL 请求——使用 ThreadPoolExecutor 并发抓取"""
from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request
import urllib.error
import time
from dataclasses import dataclass


@dataclass
class FetchResult:
    url: str
    status: int
    content_length: int
    elapsed: float
    error: str | None = None


def fetch_url(url: str, timeout: int = 10) -> FetchResult:
    """抓取单个 URL"""
    start = time.perf_counter()
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            content = resp.read()
            elapsed = time.perf_counter() - start
            return FetchResult(
                url=url,
                status=resp.status,
                content_length=len(content),
                elapsed=round(elapsed, 3),
            )
    except urllib.error.URLError as e:
        elapsed = time.perf_counter() - start
        return FetchResult(
            url=url, status=0, content_length=0,
            elapsed=round(elapsed, 3), error=str(e),
        )


def batch_fetch(urls: list[str], max_workers: int = 10) -> list[FetchResult]:
    """批量抓取 URL"""
    results = []

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

        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                result = future.result()
                results.append(result)
                status = "OK" if result.error is None else "FAIL"
                print(f"  [{status}] {url} ({result.elapsed}s)")
            except Exception as e:
                results.append(FetchResult(
                    url=url, status=0, content_length=0,
                    elapsed=0, error=str(e),
                ))

    return results


# 使用示例
if __name__ == "__main__":
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/status/404",
        "https://httpbin.org/status/500",
    ]

    start = time.perf_counter()
    results = batch_fetch(urls, max_workers=5)
    total = time.perf_counter() - start

    success = sum(1 for r in results if r.error is None)
    print(f"\n总计: {len(urls)} 个请求，成功 {success} 个，耗时 {total:.2f}s")

"""并行图像处理——使用 ProcessPoolExecutor 加速"""
from concurrent.futures import ProcessPoolExecutor, as_completed
import math
import time
from pathlib import Path


# 注意：必须在模块顶层定义函数，以便 pickle 序列化
def process_image(filepath: str) -> dict:
    """对单张图片执行灰度化 + 缩放（纯 Python 实现，无第三方依赖）"""
    # 模拟图像处理（实际项目中用 Pillow/OpenCV）
    # 这里用计算密集型操作模拟
    width, height = 1920, 1080
    pixels = width * height

    # 模拟灰度化计算
    gray_values = []
    for i in range(pixels):
        r = (i * 37) % 256
        g = (i * 73) % 256
        b = (i * 113) % 256
        gray = int(0.299 * r + 0.587 * g + 0.114 * b)
        gray_values.append(gray)

    # 模拟缩放（双线性插值）
    new_width, new_height = 960, 540
    resized = []
    for y in range(new_height):
        for x in range(new_width):
            src_x = int(x * width / new_width)
            src_y = int(y * height / new_height)
            idx = src_y * width + src_x
            resized.append(gray_values[idx] if idx < len(gray_values) else 0)

    return {
        "filepath": filepath,
        "original_size": (width, height),
        "resized_size": (new_width, new_height),
        "pixels_processed": pixels,
    }


def batch_process(image_paths: list[str], max_workers: int | None = None) -> list[dict]:
    """批量处理图片"""
    results = []

    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        future_to_path = {
            executor.submit(process_image, path): path
            for path in image_paths
        }

        for future in as_completed(future_to_path):
            path = future_to_path[future]
            try:
                result = future.result()
                results.append(result)
                print(f"  [完成] {path}")
            except Exception as e:
                print(f"  [失败] {path}: {e}")

    return results


# 使用示例
if __name__ == "__main__":
    import os

    # 模拟 8 张图片路径
    images = [f"image_{i:03d}.jpg" for i in range(8)]

    # 串行处理
    print("=== 串行处理 ===")
    start = time.perf_counter()
    serial_results = [process_image(img) for img in images]
    serial_time = time.perf_counter() - start
    print(f"串行耗时: {serial_time:.2f}s")

    # 并行处理
    print(f"\n=== 并行处理 ({os.cpu_count()} 核) ===")
    start = time.perf_counter()
    parallel_results = batch_process(images)
    parallel_time = time.perf_counter() - start
    print(f"并行耗时: {parallel_time:.2f}s")

    # 加速比
    speedup = serial_time / parallel_time
    print(f"\n加速比: {speedup:.2f}x")

"""混合 IO + CPU 任务——线程池 + 进程池协作"""
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import time
import random
import hashlib


# ---- 第一阶段：IO 密集（线程池） ----
def fetch_data(source: str) -> dict:
    """模拟从远程获取原始数据"""
    time.sleep(random.uniform(0.5, 1.5))  # 模拟网络延迟
    # 返回一段模拟数据
    raw_data = f"data_from_{source}_" + "x" * 10000
    return {"source": source, "raw_data": raw_data}


# ---- 第二阶段：CPU 密集（进程池） ----
def compute_hash(data: dict) -> dict:
    """对数据做计算密集的哈希处理"""
    raw = data["raw_data"]
    # 模拟重计算：重复哈希 10000 次
    result = raw.encode()
    for _ in range(10000):
        result = hashlib.sha256(result).digest()
    return {
        "source": data["source"],
        "hash": result.hex()[:32],
        "data_length": len(raw),
    }


def hybrid_pipeline(sources: list[str]) -> list[dict]:
    """混合流水线：线程池获取 + 进程池计算"""
    # 阶段一：线程池并发获取数据
    print(f"[阶段一] 线程池获取 {len(sources)} 个数据源...")
    fetched_data = []
    with ThreadPoolExecutor(max_workers=10) as io_pool:
        future_to_source = {
            io_pool.submit(fetch_data, src): src
            for src in sources
        }
        for future in as_completed(future_to_source):
            src = future_to_source[future]
            try:
                data = future.result()
                fetched_data.append(data)
                print(f"  [IO完成] {src}")
            except Exception as e:
                print(f"  [IO失败] {src}: {e}")

    # 阶段二：进程池并行计算
    print(f"\n[阶段二] 进程池计算 {len(fetched_data)} 个数据...")
    results = []
    with ProcessPoolExecutor() as cpu_pool:
        future_to_source = {
            cpu_pool.submit(compute_hash, data): data["source"]
            for data in fetched_data
        }
        for future in as_completed(future_to_source):
            src = future_to_source[future]
            try:
                result = future.result()
                results.append(result)
                print(f"  [计算完成] {src} -> {result['hash'][:16]}...")
            except Exception as e:
                print(f"  [计算失败] {src}: {e}")

    return results


# 使用示例
if __name__ == "__main__":
    sources = [f"api-{i}.example.com" for i in range(6)]

    start = time.perf_counter()
    results = hybrid_pipeline(sources)
    total = time.perf_counter() - start

    print(f"\n总计处理 {len(results)} 个数据源，耗时 {total:.2f}s")

from concurrent.futures import ThreadPoolExecutor

# 错误：在任务 A 中等待任务 B 的结果，而任务 B 也在等待任务 A
def task_a(future_b):
    result_b = future_b.result()  # 等待 B 完成
    return f"A got {result_b}"

def task_b(future_a):
    result_a = future_a.result()  # 等待 A 完成
    return f"B got {result_a}"

# 死锁！两个任务互相等待
with ThreadPoolExecutor(max_workers=2) as executor:
    fa = executor.submit(task_a, None)  # 需要传入 fb
    fb = executor.submit(task_b, fa)    # 需要传入 fa
    # 但 fa 需要 fb... 循环依赖 -> 死锁

from concurrent.futures import ThreadPoolExecutor, as_completed

def process(item):
    return item * 2

# 错误：一次性提交 100 万个任务，内存爆炸
# items = list(range(1_000_000))
# with ThreadPoolExecutor(max_workers=10) as executor:
#     futures = [executor.submit(process, i) for i in items]  # 内存爆炸！

# 正确：分批提交
def batch_process(items, batch_size=1000, max_workers=10):
    results = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [executor.submit(process, item) for item in batch]
            for future in as_completed(futures):
                results.append(future.result())
    return results