绿色线程

概念定义

OS 线程每条要付 MB 级栈与一个内核调度对象——扛得住数千、扛不住数百万。绿色线程把两项成本都缩进用户态：小而可增长的栈，纳秒级切换的调度器。

工作机制

运行时调度器维护可运行队列，把 M 条绿色线程复用到 N 条 OS 线程。让点是显式协作调用或编译器在函数边界插入的检查。I/O 必须走运行时感知的非阻塞接口，调度器才能挂起绿色线程而不是承载线程。

使用场景

要求“阻塞式代码可读性 + 海量并发”的场景：海量长连接、每会话状态机、流水线。代价是运行时必须接管 I/O。

常见误区

“绿色线程天然用满多核”不成立：单线程调度器会钉死在一个核上，除非运行时复用到线程池。阻塞的内核调用会冻结承载线程并连坐其上的全部绿色线程——Go 之外的旧运行时甚至从不抢占，一个忙碌任务饿死全场。共享内存的数据竞争与 OS 线程毫无二致。

代码示例

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()