异步上下文传播

概念定义

逻辑请求流经多个任务、跨越多条线程。物理执行上下文——线程——不再能标识逻辑上下文。上下文传播重定义单位：一个上下文值随异步链流动，续延恢复时无论落在哪条线程都被还原。

工作机制

机制各异：Node async_hooks 拦截 promise 与定时器回调存取存储；Python contextvars 在任务创建时拷贝映射；JVM 协程上下文流经挂起函数；Go 以显式首参传递。共同前提是续延的每一跳都经过被插桩的边界。

使用场景

扇出链路的追踪与日志关联、按租户路由存储、特性开关解析、取消截止时间。应对技术栈里每个队列与定时器边界显式测试传播。

常见误区

未插桩边界——裸线程池、调用回调的 C 库、消息队列跳转——会静默丢上下文。可变共享上下文在并发下造成跨请求串染。每个任务拷贝超大的上下文值会产生分配开销——保持精简与不可变。追踪之外的授权范围、租户路由、截止时间都依赖同一机制。

代码示例

export async function mapWithConcurrency(items, limit, mapper, options = {}) {
  if (!Array.isArray(items)) {
    throw new TypeError("items must be an array");
  }
  if (!Number.isInteger(limit) || limit < 1) {
    throw new RangeError("limit must be a positive integer");
  }
  if (typeof mapper !== "function") {
    throw new TypeError("mapper must be a function");
  }
  const { signal, stopOnError = true } = options;
  if (
    signal !== undefined &&
    (typeof signal.aborted !== "boolean" || typeof signal.addEventListener !== "function")
  ) {
    throw new TypeError("signal must be an AbortSignal");
  }

  const results = new Array(items.length);
  let nextIndex = 0;
  let failed = false;
  let firstError;

  const worker = async () => {
    while (true) {
      if (signal?.aborted) {
        throw signal.reason ?? new DOMException("Aborted", "AbortError");
      }
      if (failed && stopOnError) return;
      const index = nextIndex;
      nextIndex += 1;
      if (index >= items.length) return;
      try {
        results[index] = await mapper(items[index], index);
      } catch (error) {
        if (!failed) {
          failed = true;
          firstError = error;
        }
        if (stopOnError) return;
      }
    }
  };

  await Promise.all(
    Array.from({ length: Math.min(limit, items.length) }, () => worker()),
  );

  if (failed && stopOnError) throw firstError;
  return results;
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const order = [];
const squares = await mapWithConcurrency([1, 2, 3, 4, 5], 2, async (value) => {
  order.push(`start ${value}`);
  await sleep(value * 5);
  order.push(`end ${value}`);
  return value ** 2;
});
console.log(squares);
console.log(order.join(" | "));

const failure = await mapWithConcurrency([1, 2, 3], 2, async (value) => {
  if (value === 2) throw new Error(`failed on ${value}`);
  return value;
}).catch((error) => error.message);
console.log(failure);

export function mapPool(items, mapper, concurrency = 4) {
  if (!Array.isArray(items)) throw new TypeError("items must be an array");
  if (typeof mapper !== "function") throw new TypeError("mapper must be a function");
  if (!Number.isInteger(concurrency) || concurrency < 1) {
    throw new RangeError("concurrency must be a positive integer");
  }

  const results = new Array(items.length);
  let nextIndex = 0;

  async function worker() {
    while (nextIndex < items.length) {
      const index = nextIndex;
      nextIndex += 1;
      results[index] = await mapper(items[index], index);
    }
  }

  const workerCount = Math.min(concurrency, items.length);
  const workers = Array.from({ length: workerCount }, () => worker());
  return Promise.all(workers).then(() => results);
}