异步并发池映射工具

示例说明

全并发 map 会被下游限流，串行又太慢——有界并发是中间答案。这个实现不用队列数据结构：游标加固定工人数即可，每个工人循环领取下一个未处理索引，先完成者多干，天然负载均衡。结果写入按索引预分配的数组，输出顺序与输入严格一致，与完成顺序无关。

技术要点

• 游标领取模式免队列且自动均衡
• 结果按索引回填保序
• worker 数不超过条目数防空转

代码解读

asyncPool 按并发上限与集合长度的较小值创建工人；工人内 while 循环先取索引再调用 mapper，await 期间其他工人继续领取。第二段演示失败语义：某个条目抛错后 Promise.all 立即拒绝，剩余条目不再开始，错误消息完整传递给调用方。

使用建议

需要快速失败且取消在途任务时可加 AbortSignal 透传。生产可直接用 p-map 库而非自造。并发数与下游配额对齐，留出重试余量。

代码示例

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