线程本地存储

概念概述

线程本地存储（Thread-Local Storage）指的是：线程本地存储为每个线程提供变量副本，避免共享，但在线程池与异步调度下因线程复用而可能泄漏请求间状态。

从更严格的定义看，Thread-local storage is a mechanism that allocates a distinct instance of a variable per thread, so each thread reads and writes its own copy without synchronisation. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。

为什么需要它

It is used for per-request context, buffers, and statistics, but requires care when work migrates between threads. 它与 Asynchronous I/O、Goroutine and Task Leaks、I/O Multiplexing with epoll and kqueue 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。

常见误解

• thread-local values are cleaned up automatically when a request ends
• thread-local storage is a substitute for passing context explicitly

需要避免的反模式

• storing request identity in thread-local state behind a thread pool
• assuming async continuations resume on the same thread

实践要点

1. Because a pooled thread is reused for many requests, a stale value left in thread-local storage can leak from one request to the next, a subtle data isolation bug.
2. In async and coroutine runtimes the logical task may resume on a different thread, so context must follow the task rather than the thread.
3. Prefer explicit propagation of context, or a runtime-provided context object, and if thread-local storage is used, always reset it in a finally block.

参考来源

https://en.wikipedia.org/wiki/Thread-local_storage

代码示例

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

package main

import (
	"context"
	"fmt"
	"net/http"

	"golang.org/x/sync/errgroup"
)

type Result struct {
	URL  string
	Code int
}

func fetchAll(ctx context.Context, urls []string) ([]Result, error) {
	g, ctx := errgroup.WithContext(ctx)
	results := make([]Result, len(urls))

	for i, url := range urls {
		i, url := i, url // 捕获循环变量（Go 1.22 前必需）
		g.Go(func() error {
			req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
			if err != nil {
				return fmt.Errorf("build request %s: %w", url, err)
			}
			resp, err := http.DefaultClient.Do(req)
			if err != nil {
				return fmt.Errorf("fetch %s: %w", url, err)
			}
			defer resp.Body.Close()
			results[i] = Result{URL: url, Code: resp.StatusCode}
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err // ctx 已取消，其余 goroutine 全部提前退出
	}
	return results, nil
}

func main() {
	rs, err := fetchAll(
		context.Background(),
		[]string{"https://example.com/a", "https://example.com/b"},
	)
	fmt.Println(len(rs), err)
}