线程池

概念定义

线程池在启动时创建固定或受限数量的工作线程。任务进入队列；空闲的工作线程取走、执行，再回来取下一个。当所有工作线程都在忙时任务等待，当队列也满时，线程池必须拒绝、阻塞或施加背压。于是线程池把开放式的需求转化为有界、可调的并发供给，并复用那些否则要按任务付出创建成本的线程。

为什么重要

为每个请求创建一个线程会把内存与 CPU 浪费在调度上，而无界并发会让一次尖峰压垮进程。线程池限定资源使用、平滑突发，并让在途任务数量与机器实际能处理的量保持一致。它也是表达过载策略的层次：排队、卸载或减速。

工程实践

按瓶颈给线程池定尺寸：CPU 密集型大致等于核心数，I/O 密集型因等待时间而更高，并以测量而非公式来验证。保持任务彼此独立，避免工作线程等待同一池中的另一个任务，并设置有界队列与明确的拒绝或背压策略。监控队列深度与活跃工作线程，因为队列增长是饱和最早的信号。

代码示例

import java.util.concurrent.*;
import java.util.List;

public class ThreadPoolExecutor {
    public static void main(String[] args) throws InterruptedException {
        try (ExecutorService executor = Executors.newFixedThreadPool(4)) {
            List<Callable<String>> tasks = List.of(
                () -> { Thread.sleep(500); return "Task 1 done"; },
                () -> { Thread.sleep(300); return "Task 2 done"; },
                () -> { Thread.sleep(700); return "Task 3 done"; },
                () -> { Thread.sleep(200); return "Task 4 done"; }
            );

            List<Future<String>> futures = executor.invokeAll(tasks);

            for (Future<String> future : futures) {
                try {
                    System.out.println(future.get());
                } catch (ExecutionException e) {
                    System.err.println("Task failed: " + e.getCause());
                }
            }
        }
        System.out.println("All tasks completed");
    }
}

import time
from collections.abc import Callable, Sequence
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from concurrent.futures import as_completed
from typing import TypeVar

T = TypeVar("T")
R = TypeVar("R")


def pool_map(
    func: Callable[[T], R],
    items: Sequence[T],
    *,
    workers: int = 4,
    timeout: float | None = None,
) -> list[R]:
    if workers < 1:
        raise ValueError("workers must be positive")
    if not items:
        return []
    results: dict[int, R] = {}
    with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="map") as pool:
        futures = {pool.submit(func, item): index for index, item in enumerate(items)}
        try:
            for future in as_completed(futures, timeout=timeout):
                results[futures[future]] = future.result()
        except FutureTimeout:
            for future in futures:
                future.cancel()
            raise
    return [results[index] for index in range(len(items))]


def main() -> None:
    def slow_square(value: int) -> int:
        time.sleep(0.05)
        return value * value

    numbers = list(range(1, 7))
    start = time.perf_counter()
    squares = pool_map(slow_square, numbers, workers=3)
    print("squares:", squares)
    print(f"elapsed: {time.perf_counter() - start:.2f}s")

    try:
        pool_map(lambda value: time.sleep(0.3) or value, [1, 2], timeout=0.05)
    except FutureTimeout:
        print("timed out and cancelled pending work")


if __name__ == "__main__":
    main()