结构化并发

概念定义

代码块约束资源生命周期，而传统并发不约束任务。游离的 promise 是无主资源：没人取消它、没人看它的错误、没人限定它的寿命。结构化并发给任务与局部变量同样的生命周期规则。

工作机制

作用域开启、子任务注册、汇合点强制完成语义：等全部、首个成功即取消其余、或聚合结果。失败的子任务把作用域标记为失败并请求取消兄弟；汇合点重抛代表性错误。作用域退出保证没有任务活得比代码块更久。

使用场景

向多个服务扇出的请求处理器、带早退条件的批任务、以及“部分失败就该全停”的场景。要与取消令牌或协作让点配合，子任务才会真正响应作用域取消。

常见误区

“等价于 Promise.all”低估了它：all 只等结果，既无子任务取消也无未决拒绝的兜底。在忽略取消的子任务里写阻塞，会把结构化纪律变成挂死。自管生命周期的遗留 API（定时器、watcher）需要适配器注册进作用域；真正的后台工作确需逃逸舱，但必须显式、可见、可评审。

代码示例

interface RetryableError extends Error {
  retryable?: boolean;
}

class PermanentError extends Error {
  retryable = false;
}

function classify(err: unknown): RetryableError {
  if (err instanceof PermanentError) return err;

  const message = err instanceof Error ? err.message : String(err);
  const transient =
    /timeout|econnreset|socket hang up|503|429/i.test(message);

  return Object.assign(
    err instanceof Error ? err : new Error(message),
    { retryable: transient },
  );
}

async function withRetry<T>(
  fn: () => Promise<T>,
  { attempts = 3, baseMs = 200 } = {},
): Promise<T> {
  let lastError: RetryableError;

  for (let i = 1; i <= attempts; i++) {
    try {
      return await fn();
    } catch (raw) {
      lastError = classify(raw);

      if (!lastError.retryable || i === attempts) {
        throw lastError;
      }
      await new Promise((r) => setTimeout(r, baseMs * 2 ** (i - 1)));
    }
  }
  throw lastError!;
}

const result = await withRetry(async () => {
  if (Math.random() < 0.7) {
    const err = new Error('upstream 503 unavailable');
    throw err;
  }
  return 'settled';
});
console.log('outcome:', result);

async function asyncPool<T, R>(
  concurrency: number,
  items: readonly T[],
  mapper: (item: T, index: number) => Promise<R>,
): Promise<R[]> {
  const results = new Array<R>(items.length);
  let cursor = 0;

  const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {
    while (cursor < items.length) {
      const index = cursor;
      const item = items[index];
      cursor += 1;
      results[index] = await mapper(item, index);
    }
  });

  await Promise.all(workers);
  return results;
}

const CONCURRENCY = 4;

const enriched = await asyncPool(CONCURRENCY, ['a', 'b', 'c', 'd', 'e', 'f'], async (id, i) => {
  await new Promise((r) => setTimeout(r, 20));
  return { id, position: i };
});

console.log('order preserved:', enriched.map((e) => e.id).join(''));

const withFailure = await asyncPool(2, ['x', 'y', 'z'], async (id) => {
  if (id === 'y') throw new Error('boom on ' + id);
  return id.toUpperCase();
}).catch((err) => err.message);

console.log('failure propagated:', withFailure);