JavaScript 异步并发控制完全指南：手写 Promise 调度器与生产级并发池

2026 年，一个中等规模的前端项目动辄需要同时调用 10-20 个 API 接口，而浏览器对同一域名的并发连接数限制在 6 个左右。如果你直接用 Promise.all() 打出去 20 个请求，其中 14 个会被排队等待，加上每个请求 200ms 的延迟，用户要等将近 1 秒才能看到完整页面——这就是为什么异步并发控制是每个 JavaScript 开发者必须掌握的核心技能。

🔧 一、为什么需要并发控制？

1.1 浏览器与 Node.js 的并发瓶颈

浏览器对同一域名的 HTTP/1.1 连接数限制为 6 个（HTTP/2 下是单连接多路复用，但流的优先级调度依然存在）。Node.js 虽然没有连接数限制，但底层的 libuv 线程池默认只有 4 个线程，DNS 解析、文件 I/O 等操作都会竞争这些线程。

⚠️ 警告：Promise.all() 不是万能解药。它会同时发起所有 Promise，当数量超过系统承受能力时，会导致请求超时、内存飙升甚至被服务端限流（Rate Limit）。

下面是一个典型的反面案例：

// ❌ 错误写法：一次性发起 100 个请求，大概率被限流或超时
const urls = Array.from({ length: 100 }, (_, i) => `/api/user/${i}`);
const results = await Promise.all(urls.map(url => fetch(url)));

1.2 并发控制的本质

并发控制的核心思想很简单：维护一个固定大小的执行槽（Slot），只有当前槽位空出来时，才从等待队列中取出下一个任务执行。

这和餐厅的服务逻辑一样——餐厅只有 6 张桌子（并发数），第 7 桌及以后的客人必须在门口等位。并发控制就是这个「等位系统」。

🚀 二、从零实现并发调度器

2.1 基础版：最简并发控制器

我们先实现一个最简单的并发控制器，核心逻辑只有 20 行：

// 并发控制器基础版：限制同时执行的 Promise 数量
class ConcurrencyController {
  constructor(concurrency = 6) {
    this.concurrency = concurrency;    // 最大并发数
    this.running = 0;                  // 当前正在执行的任务数
    this.queue = [];                   // 等待队列
  }

  // 提交一个异步任务
  async submit(asyncFn) {
    // 如果当前并发数已满，等待槽位释放
    if (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }

    this.running++;
    try {
      return await asyncFn();
    } finally {
      this.running--;
      // 从队列中取出下一个等待的任务，释放槽位
      if (this.queue.length > 0) {
        const next = this.queue.shift();
        next();
      }
    }
  }
}

// 使用示例：100 个请求，最多同时执行 6 个
const controller = new ConcurrencyController(6);
const urls = Array.from({ length: 100 }, (_, i) => `/api/user/${i}`);

const results = await Promise.all(
  urls.map(url => controller.submit(() => fetch(url).then(r => r.json())))
);

⚡ **关键结论：**这个实现的核心技巧是用 new Promise 创建一个「信号量」——当并发数满时，submit 方法会阻塞在 await 上，直到某个任务完成后调用 next() 释放等待。

2.2 增强版：带超时、重试和优先级的并发池

生产环境需要更多功能——超时控制、失败重试、任务优先级。下面是完整实现：

// 生产级并发池：支持超时、重试、优先级
class AsyncPool {
  constructor(options = {}) {
    this.concurrency = options.concurrency ?? 6;
    this.timeout = options.timeout ?? 30000;       // 默认 30 秒超时
    this.retries = options.retries ?? 0;            // 默认不重试
    this.running = 0;
    this.queue = [];  // 按优先级排序的等待队列
  }

  async submit(asyncFn, options = {}) {
    const priority = options.priority ?? 0;
    const retries = options.retries ?? this.retries;

    // 等待槽位
    if (this.running >= this.concurrency) {
      await new Promise(resolve => {
        this.queue.push({ resolve, priority });
        this.queue.sort((a, b) => b.priority - a.priority); // 高优先级排前面
      });
    }

    this.running++;
    let lastError;

    for (let attempt = 0; attempt <= retries; attempt++) {
      try {
        // 带超时的 Promise.race
        const result = await Promise.race([
          asyncFn(),
          new Promise((_, reject) =>
            setTimeout(() => reject(new Error('Task timeout')), this.timeout)
          ),
        ]);
        return result;
      } catch (err) {
        lastError = err;
        if (attempt < retries) {
          // 指数退避重试：100ms, 200ms, 400ms...
          await new Promise(r => setTimeout(r, 100 * Math.pow(2, attempt)));
        }
      } finally {
        // 只在最后一次尝试后释放槽位
        if (attempt === retries) {
          this.running--;
          if (this.queue.length > 0) {
            this.queue.shift().resolve();
          }
        }
      }
    }

    throw lastError;
  }

  // 批量提交，返回所有结果（类似 Promise.allSettled）
  async all(tasks) {
    return Promise.allSettled(
      tasks.map(({ fn, ...opts }) => this.submit(fn, opts))
    );
  }
}

// 使用示例
const pool = new AsyncPool({ concurrency: 4, timeout: 5000, retries: 2 });

const results = await pool.all([
  { fn: () => fetch('/api/critical').then(r => r.json()), priority: 10 },
  { fn: () => fetch('/api/normal-1').then(r => r.json()), priority: 1 },
  { fn: () => fetch('/api/normal-2').then(r => r.json()), priority: 1 },
  { fn: () => fetch('/api/low').then(r => r.json()), priority: 0 },
]);

results.forEach((result, i) => {
  if (result.status === 'fulfilled') {
    console.log(`✅ 任务 ${i} 成功:`, result.value);
  } else {
    console.log(`❌ 任务 ${i} 失败:`, result.reason.message);
  }
});

2.3 Generator 版：利用迭代器实现惰性调度

Generator 提供了一种更优雅的写法，特别适合处理大量数据的分批并发：

// Generator 版并发控制器：惰性调度，内存友好
async function* asyncPool(concurrency, items, asyncFn) {
  const executing = new Set();

  for (const item of items) {
    // 创建一个 Promise 并立即开始执行
    const p = Promise.resolve().then(() => asyncFn(item));
    executing.add(p);

    // 清理完成的 Promise
    const clean = () => executing.delete(p);
    p.then(clean, clean);

    // 当并发数达到上限时，等待最快完成的那个
    if (executing.size >= concurrency) {
      yield await Promise.race(executing);
    }
  }

  // 等待剩余的任务全部完成
  while (executing.size > 0) {
    yield await Promise.race(executing);
  }
}

// 使用示例：分批获取 1000 条数据，每次最多 5 个并发
const userIds = Array.from({ length: 1000 }, (_, i) => i);

for await (const userData of asyncPool(5, userIds, async (id) => {
  const res = await fetch(`/api/user/${id}`);
  return res.json();
})) {
  // 每完成一个就立即处理，不需要等所有任务完成
  console.log('Got user:', userData.name);
}

💡 提示：Generator 版的优势在于惰性求值——不需要一次性创建 1000 个 Promise，而是在运行时动态创建，对内存更加友好。

📊 三、主流库横向对比

社区已经有很多成熟的并发控制库，下面是主流方案的详细对比：

3.1 p-limit：最流行的轻量方案

p-limit 是 npm 上周下载量超过 3000 万 的并发控制库，API 设计极简：

import pLimit from 'p-limit';

const limit = pLimit(6); // 最大并发数 6

const urls = Array.from({ length: 100 }, (_, i) => `/api/data/${i}`);

// 所有请求共享同一个并发池
const results = await Promise.all(
  urls.map(url => limit(() => fetch(url).then(r => r.json())))
);

console.log(`并发数: ${limit.activeCount}`);   // 当前正在执行的数量
console.log(`等待数: ${limit.pendingCount}`);  // 队列中等待的数量

📌 记住：p-limit 的核心实现就是我们上面手写的「基础版并发控制器」，源码不到 50 行。理解了原理，你就能在任何语言中实现相同的逻辑。

3.2 p-queue：功能最全的队列方案

当需要优先级、暂停/恢复、事件监听等高级功能时，p-queue 是更好的选择：

import PQueue from 'p-queue';

const queue = new PQueue({
  concurrency: 4,
  interval: 1000,        // 每秒最多执行
  intervalCap: 10,       // 每秒最多 10 个任务
});

// 监听队列事件
queue.on('active', () => console.log(`活跃任务: ${queue.size}`));
queue.on('idle', () => console.log('队列已清空'));

// 添加带优先级的任务
await queue.add(() => fetch('/api/urgent'), { priority: 10 });
await queue.add(() => fetch('/api/normal'), { priority: 1 });

// 暂停队列（用于背压控制）
queue.pause();
// ... 稍后恢复
queue.start();

3.3 性能基准测试

我对这三种方案做了基准测试（1000 个模拟任务，每个耗时 10-50ms 随机延迟，最大并发 10）：

⚡ 关键结论：性能差异可以忽略不计（都在 4% 以内）。选型的关键标准是功能需求——简单场景用 p-limit，需要队列管理用 p-queue，追求极致内存用 Generator 版。

💡 四、实战场景与避坑指南

4.1 场景一：批量文件上传

电商后台的图片批量上传是一个典型场景。前端需要同时上传 50 张图片到 OSS，每张图片 2-5MB：

// 批量上传：并发数 3，带进度回调
import pLimit from 'p-limit';

const limit = pLimit(3);

async function batchUpload(files, onProgress) {
  let completed = 0;
  const total = files.length;

  const tasks = files.map(file =>
    limit(async () => {
      const formData = new FormData();
      formData.append('file', file);

      const response = await fetch('/api/upload', {
        method: 'POST',
        body: formData,
      });

      completed++;
      onProgress({ completed, total, percent: Math.round(completed / total * 100) });

      return response.json();
    })
  );

  return Promise.allSettled(tasks);
}

// 使用
const results = await batchUpload(selectedFiles, ({ percent }) => {
  console.log(`上传进度: ${percent}%`);
});

const failed = results.filter(r => r.status === 'rejected');
if (failed.length > 0) {
  console.warn(`⚠️ ${failed.length} 个文件上传失败`);
}

⚠️ **警告：**上传场景的并发数不要设太高！浏览器的 TCP 连接数有限，加上每个上传请求体很大，太高并发会导致所有请求都变慢。建议设为 3-4。

4.2 场景二：API 限流下的并发请求

很多 API 都有速率限制（如 GitHub API 每分钟 60 次，OpenAI API 每分钟 60 次）。这时候需要「速率限制器」而非简单的并发限制器：

// 令牌桶限流器：每秒最多 N 个请求
class TokenBucketRateLimiter {
  constructor(tokensPerInterval, intervalMs = 1000) {
    this.tokensPerInterval = tokensPerInterval;
    this.intervalMs = intervalMs;
    this.tokens = tokensPerInterval;
    this.lastRefill = Date.now();
    this.queue = [];
  }

  refill() {
    const now = Date.now();
    const elapsed = now - this.lastRefill;
    const refillCount = (elapsed / this.intervalMs) * this.tokensPerInterval;
    this.tokens = Math.min(this.tokensPerInterval, this.tokens + refillCount);
    this.lastRefill = now;
  }

  async acquire() {
    this.refill();

    if (this.tokens >= 1) {
      this.tokens -= 1;
      return;
    }

    // 计算需要等待的时间
    const waitMs = ((1 - this.tokens) / this.tokensPerInterval) * this.intervalMs;
    await new Promise(resolve => setTimeout(resolve, waitMs));
    this.tokens = 0;
    this.lastRefill = Date.now();
  }
}

// 使用：GitHub API 限流（每分钟 60 次 = 每秒 1 次）
const limiter = new TokenBucketRateLimiter(1, 1000);

const repos = ['vue', 'react', 'angular', 'svelte', 'solid'];
const results = await Promise.all(
  repos.map(async repo => {
    await limiter.acquire(); // 自动等待直到可以发送请求
    return fetch(`https://api.github.com/repos/${repo}`).then(r => r.json());
  })
);

4.3 常见坑点汇总

🎯 总结与建议

经过上面的分析和实战，以下是并发控制的最佳实践：

选型建议：

• ✅ 简单场景（批量 API 调用、图片上传）→ 直接用 p-limit，1.2KB 完事
• ✅ 复杂场景（优先级队列、暂停恢复、速率限制）→ 用 p-queue 或 bottleneck
• ✅ 超大批量（10 万+ 任务）→ 用 Generator 版惰性调度，内存最优
• ✅ API 限流（GitHub、OpenAI 等）→ 用令牌桶（Token Bucket）算法
• ❌ 避免直接用 Promise.all() 处理超过 10 个并发请求

通用配置建议：

• 浏览器端 API 调用：并发数 4-6
• 文件上传：并发数 2-3
• Node.js 后端调用外部 API：并发数 10-20（视 API 限流而定）
• 数据库批量操作：并发数 5-10

💡 **提示：**并发控制的本质是一个「资源管理」问题。核心变量永远是三个：并发数、超时时间、重试策略。把这三个参数配对了，90% 的并发问题都能解决。

相关工具推荐：

• 🔧 p-limit — 极简并发限制器
• 🔧 p-queue — 功能齐全的优先级队列
• 🔧 bottleneck — 专业的 API 限流器
• 🔧 tiny-async-pool — 极轻量方案（0.5KB）
• 🔧 jsjson.com/json-format — 在线 JSON 格式化工具，调试 API 响应时必备