异步并发控制

1. 学习目标（Bloom 分类）

读完本文后，读者应能够达到以下认知层次：

学习本课前，建议先掌握：Promise、async/await、事件循环、闭包、Class、WeakMap。

2. 历史动机：为什么需要并发控制

2.1 浏览器与 Node.js 的并发约束

JavaScript 是单线程语言，但其异步 I/O 能力使其在网络密集型任务中表现卓越。然而，“单线程”并不等于”无限并发”：

• 浏览器连接数限制：HTTP/1.1 规范建议浏览器对同一域名最多保持 6 个 TCP 连接（Chrome、Firefox 实际为 6，Safari 为 4）。超过限制的请求会被排队等待。
• 服务器限流：API 服务通常设置 QPS（Queries Per Second）阈值，超限返回 429 Too Many Requests。
• 内存压力：每个 Promise 对象、请求上下文、响应缓冲都占用内存。1 万个并发请求可能瞬间耗尽 V8 堆。
• 文件描述符耗尽：Node.js 进程默认可打开 1024 个文件描述符（可通过 ulimit -n 调整），并发请求过多会触发 EMFILE 错误。
• CPU 竞争：虽然 I/O 是异步的，但 JSON 解析、加密、压缩等 CPU 密集操作仍会争抢主线程时间。

2.2 无限制并发的灾难场景

// 反模式：一次性发起 10000 个请求
async function fetchAllBad(urls) {
  return Promise.all(urls.map((url) => fetch(url)));
}

const urls = Array.from({ length: 10000 }, (_, i) => `/api/item/${i}`);
await fetchAllBad(urls);
// 实际后果：
// 1. 浏览器排队 9994 个请求，等待 6 个连接释放
// 2. 服务器可能因瞬时 QPS 超限返回 429
// 3. 浏览器标签页内存暴涨，可能崩溃
// 4. 单个请求超时会触发连锁重试，加剧雪崩

2.3 p-limit 的诞生与流行

2017 年，Sindre Sorhus 发布了 p-limit 库，提供简洁的 API：

import pLimit from 'p-limit';

const limit = pLimit(3);
const results = await Promise.all(
  urls.map((url) => limit(() => fetch(url)))
);

其核心思想是：用一个共享的”运行槽”计数器，让超过并发度的任务在内部队列中排队。p-limit 至今仍是 npm 上最流行的并发控制库（周下载量超过 2 亿次）。

2.4 现代前端/Node.js 的并发控制需求

2.5 商业价值

并发控制不只是技术细节，更是系统稳定性的关键：

• 避免雪崩：Netflix 的 Hystrix 库通过并发限制与断路器，将故障隔离从”全盘崩溃”降到”局部降级”。
• 降低成本：云服务按请求计费，合理的并发度可避免突发流量导致的计费峰值。
• 提升 SLA：通过限流保护核心服务，确保关键路径的可用性。
• 简化运维：固定并发度让容量规划变得可预测，避免动态扩容的复杂性。

3. 形式化定义

3.1 并发控制的核心概念

定义并发控制系统 $C$ 为五元组：

其中：

• $T = \{t_1, t_2, ..., t_n\}$：任务集合，每个任务 $t_i$ 是一个返回 Promise 的函数。
• $Q$：等待队列，存储尚未启动的任务。
• $R$：运行集合，存储正在执行的任务（满足 $|R| \leq \tau$）。
• $\tau$：最大并发度（concurrency limit），正整数。
• $\sigma$：调度策略（FIFO、LIFO、优先级、公平轮询）。

3.2 并发控制的不变量

并发控制必须始终维持以下不变量：

违反不变量 1 会导致并发度失控，违反不变量 2 会导致吞吐量下降，违反不变量 3 会导致队列停滞。

3.3 吞吐量与延迟的形式化

定义：

• $T_p$：单个任务平均执行时间。
• $\tau$：并发度。
• $T_{total}$：总执行时间。
• $n$：任务总数。

理想吞吐量（无队列等待）：

总执行时间下界：

单个任务的平均延迟：

其中 $W_{avg}$ 是平均队列等待时间。当 $\tau \geq n$ 时 $W_{avg} = 0$；当 $\tau < n$ 时 $W_{avg}$ 随任务位置递增。

3.4 利特尔定律（Little’s Law）

并发控制系统符合排队论的利特尔定律：

其中：

• $L$：系统中平均任务数（包括排队和执行）。
• $\lambda$：任务到达速率。
• $W$：平均逗留时间。

在稳定状态下，并发度 $\tau$ 即系统的”服务台数量”。当 $\lambda > \tau / T_p$ 时，队列会无限增长，系统不稳定。

3.5 调度策略的形式化

FIFO（先进先出）：

LIFO（后进先出）：

优先级调度：

公平轮询：

其中 $w_c$ 是类别 $c$ 的权重。

4. 理论推导

4.1 从回调到 Promise 的演进

回调时代（2010 年前）：

function fetchOne(url, callback) {
  setTimeout(() => callback(null, `data:${url}`), 100);
}

// 串行
function serial(urls, cb) {
  const results = [];
  let i = 0;
  function next(err, data) {
    if (err) return cb(err);
    results.push(data);
    i++;
    if (i < urls.length) {
      fetchOne(urls[i], next);
    } else {
      cb(null, results);
    }
  }
  fetchOne(urls[i], next);
}

回调时代的并发控制需要手动管理计数器，代码极易出错。

Promise 时代（ES6, 2015）：

function fetchOne(url) {
  return new Promise((resolve) => {
    setTimeout(() => resolve(`data:${url}`), 100);
  });
}

// 无并发限制
Promise.all(urls.map(fetchOne)).then((results) => {
  console.log(results);
});

Promise.all 解决了”等待全部完成”的问题，但未解决”同时启动”的问题——map 会立即创建所有 Promise，所有任务”同时”启动。

async/await 时代（ES2017）：

async function serial(urls) {
  const results = [];
  for (const url of urls) {
    results.push(await fetchOne(url));
  }
  return results;
}

for...of + await 是严格的串行，性能极差。要并发又要有控制，必须引入并发限制器。

4.2 p-limit 的核心算法

p-limit 的核心是一个闭包 + 队列：

function pLimit(concurrency) {
  // 不变量 1：activeCount 始终 <= concurrency
  let activeCount = 0;
  const queue = [];

  const next = () => {
    activeCount--;
    // 不变量 2：有空闲槽且有任务时启动下一个
    if (queue.length > 0) {
      queue.shift()();
    }
  };

  const run = async (fn, resolve, reject) => {
    activeCount++;
    try {
      const result = await fn();
      resolve(result);
    } catch (error) {
      reject(error);
    } finally {
      next(); // 不变量 3：任务完成时触发调度
    }
  };

  const enqueue = (fn) => {
    return new Promise((resolve, reject) => {
      if (activeCount < concurrency) {
        run(fn, resolve, reject);
      } else {
        queue.push(() => run(fn, resolve, reject));
      }
    });
  };

  return enqueue;
}

算法分析：

• 时间复杂度：每个任务入队、出队均为 $O(1)$（数组 shift 是 $O(n)$，可用链表优化为 $O(1)$）。
• 空间复杂度：$O(n)$（队列存储所有待执行任务）。
• 公平性：FIFO，先入先出。

4.3 Promise.all + map 的并发陷阱

// 反模式
const limit = pLimit(3);
const promises = urls.map((url) => limit(() => fetch(url)));
// 此时所有 limit() 已被调用，所有任务都已入队
const results = await Promise.all(promises);

注意：map 会同步遍历整个数组，所有任务在 Promise.all 执行前就已入队。这意味着：

• 队列立即有 $n$ 个任务。
• p-limit 内部按 FIFO 顺序启动，最多 3 个并发。
• Promise.all 等待所有 Promise resolve。

这是正确用法。但要注意：所有 Promise 在 map 时就已创建，意味着任务的”创建时机”和”执行时机”是分离的。如果任务函数有副作用（如读取当前时间），需特别注意。

4.4 队列调度的微任务语义

p-limit 的 next() 函数在 finally 中调用，意味着它在任务完成后立即触发下一个任务。但 JavaScript 的事件循环规则：

• await fn() 后的代码在微任务中执行。
• queue.shift()() 同步调用 run()，但 run 内部的 await fn() 让出执行权。
• 下一个任务启动后，控制权回到 Promise.all 的等待链。

这意味着并发控制不会引入额外的宏任务延迟，性能接近最优。

4.5 吞吐量推导

假设：

• $n$ 个任务，每个执行时间 $T_p$。
• 并发度 $\tau$。
• 调度开销 $T_s$（可忽略）。

总执行时间：

吞吐量：

加速比（相对串行）：

Amdahl 定律的修正：

经典 Amdahl 定律假设并行部分可任意拆分。在并发控制中，受限于 $\tau$，加速比上限为 $\tau$：

4.6 队列长度的稳定性分析

设任务到达速率为 $\lambda$（任务/秒），服务速率为 $\mu = \tau / T_p$（任务/秒）。

M/D/1 排队模型（确定服务时间）：

• 稳定条件：$\lambda < \mu$。
• 平均队列长度：$L_q = \frac{\rho^2}{2(1-\rho)}$，其中 $\rho = \lambda / \mu$。
• 平均等待时间：$W_q = \frac{\rho T_p}{2(1-\rho)}$。

当 $\rho \to 1$（即 $\lambda \to \mu$）时，队列长度趋向无穷大。因此并发度必须留有余量，通常设置 $\rho < 0.7$ 以保证稳定性。

5. 代码示例

5.1 基础 p-limit 实现

/**
 * 基础并发限制器
 * @param {number} concurrency - 最大并发度
 * @returns {Function} limit 函数
 */
function pLimit(concurrency) {
  if (concurrency < 1 || !Number.isInteger(concurrency)) {
    throw new TypeError('concurrency 必须是正整数');
  }

  let activeCount = 0;
  const queue = [];

  const next = () => {
    activeCount--;
    if (queue.length > 0) {
      queue.shift()();
    }
  };

  const run = async (fn, resolve, reject) => {
    activeCount++;
    try {
      const result = await fn();
      resolve(result);
    } catch (error) {
      reject(error);
    } finally {
      next();
    }
  };

  const enqueue = (fn) => {
    return new Promise((resolve, reject) => {
      if (activeCount < concurrency) {
        run(fn, resolve, reject);
      } else {
        queue.push(() => run(fn, resolve, reject));
      }
    });
  };

  // 暴露状态查询
  Object.defineProperty(enqueue, 'activeCount', {
    get: () => activeCount,
  });
  Object.defineProperty(enqueue, 'pendingCount', {
    get: () => queue.length,
  });

  return enqueue;
}

// 使用示例
const limit = pLimit(3);
const urls = Array.from({ length: 10 }, (_, i) => `/api/${i}`);

async function mockFetch(url) {
  console.log(`[start] ${url}`);
  await new Promise((r) => setTimeout(r, 100 + Math.random() * 200));
  console.log(`[done] ${url}`);
  return `data:${url}`;
}

const results = await Promise.all(
  urls.map((url) => limit(() => mockFetch(url)))
);
console.log(results);

5.2 使用链表优化队列性能

Array.prototype.shift() 是 $O(n)$ 操作，对大队列性能有影响。改用链表：

class Node {
  constructor(value) {
    this.value = value;
    this.next = null;
  }
}

class LinkedList {
  constructor() {
    this.head = null;
    this.tail = null;
    this.size = 0;
  }

  push(value) {
    const node = new Node(value);
    if (this.tail) {
      this.tail.next = node;
    } else {
      this.head = node;
    }
    this.tail = node;
    this.size++;
  }

  shift() {
    if (!this.head) return undefined;
    const value = this.head.value;
    this.head = this.head.next;
    if (!this.head) this.tail = null;
    this.size--;
    return value;
  }

  get length() {
    return this.size;
  }
}

function pLimitFast(concurrency) {
  let activeCount = 0;
  const queue = new LinkedList();

  const next = () => {
    activeCount--;
    if (queue.length > 0) {
      const task = queue.shift();
      task();
    }
  };

  const run = async (fn, resolve, reject) => {
    activeCount++;
    try {
      const result = await fn();
      resolve(result);
    } catch (error) {
      reject(error);
    } finally {
      next();
    }
  };

  const enqueue = (fn) => {
    return new Promise((resolve, reject) => {
      if (activeCount < concurrency) {
        run(fn, resolve, reject);
      } else {
        queue.push(() => run(fn, resolve, reject));
      }
    });
  };

  return enqueue;
}

性能对比（10 万任务，并发度 10）：

5.3 完整的 AsyncQueue 类

/**
 * 异步任务队列
 * 支持优先级、取消、暂停、统计
 */
class AsyncQueue {
  constructor(options = {}) {
    this.concurrency = options.concurrency || 4;
    this.running = 0;
    this.queue = [];
    this.paused = false;
    this.stats = {
      completed: 0,
      failed: 0,
      totalWaitTime: 0,
      totalRunTime: 0,
    };
    this.taskId = 0;
  }

  /**
   * 入队任务
   * @param {Function} task - 返回 Promise 的函数
   * @param {Object} options - { priority, id }
   * @returns {Promise} 任务结果
   */
  push(task, options = {}) {
    const id = options.id ?? ++this.taskId;
    const priority = options.priority ?? 0;
    const enqueuedAt = Date.now();

    return new Promise((resolve, reject) => {
      const item = {
        id,
        task,
        resolve,
        reject,
        priority,
        enqueuedAt,
        startedAt: null,
      };

      // 按优先级插入（高优先级在前，同优先级 FIFO）
      const insertIndex = this.queue.findIndex((q) => q.priority < priority);
      if (insertIndex === -1) {
        this.queue.push(item);
      } else {
        this.queue.splice(insertIndex, 0, item);
      }

      this.run();
    });
  }

  /**
   * 取消任务
   * @param {number} id - 任务 ID
   * @returns {boolean} 是否取消成功
   */
  cancel(id) {
    const index = this.queue.findIndex((q) => q.id === id);
    if (index !== -1) {
      const [item] = this.queue.splice(index, 1);
      item.reject(new Error('Task cancelled'));
      return true;
    }
    return false;
  }

  /**
   * 清空队列
   */
  clear() {
    const items = this.queue.splice(0);
    for (const item of items) {
      item.reject(new Error('Queue cleared'));
    }
  }

  /**
   * 暂停调度
   */
  pause() {
    this.paused = true;
  }

  /**
   * 恢复调度
   */
  resume() {
    this.paused = false;
    this.run();
  }

  /**
   * 调度循环
   */
  run() {
    if (this.paused) return;

    while (this.running < this.concurrency && this.queue.length > 0) {
      const item = this.queue.shift();
      this.running++;
      item.startedAt = Date.now();

      const waitTime = item.startedAt - item.enqueuedAt;
      this.stats.totalWaitTime += waitTime;

      Promise.resolve()
        .then(() => item.task())
        .then(
          (result) => {
            const runTime = Date.now() - item.startedAt;
            this.stats.totalRunTime += runTime;
            this.stats.completed++;
            item.resolve(result);
          },
          (error) => {
            this.stats.failed++;
            item.reject(error);
          }
        )
        .finally(() => {
          this.running--;
          this.run();
        });
    }
  }

  /**
   * 获取队列状态
   */
  get status() {
    const avgWait = this.stats.completed > 0
      ? this.stats.totalWaitTime / (this.stats.completed + this.stats.failed)
      : 0;
    const avgRun = this.stats.completed > 0
      ? this.stats.totalRunTime / (this.stats.completed + this.stats.failed)
      : 0;

    return {
      running: this.running,
      pending: this.queue.length,
      completed: this.stats.completed,
      failed: this.stats.failed,
      paused: this.paused,
      avgWaitTime: avgWait,
      avgRunTime: avgRun,
    };
  }

  /**
   * 等待队列空闲
   * @param {number} timeout - 超时（毫秒）
   */
  async idle(timeout = 0) {
    const start = Date.now();
    return new Promise((resolve, reject) => {
      const check = () => {
        if (this.running === 0 && this.queue.length === 0) {
          resolve();
        } else if (timeout > 0 && Date.now() - start > timeout) {
          reject(new Error('Idle timeout'));
        } else {
          setTimeout(check, 50);
        }
      };
      check();
    });
  }
}

// 使用示例
const queue = new AsyncQueue({ concurrency: 3 });

async function processItem(item) {
  const wait = 100 + Math.random() * 200;
  await new Promise((r) => setTimeout(r, wait));
  return `processed:${item}`;
}

async function main() {
  const items = Array.from({ length: 20 }, (_, i) => i);

  // 高优先级任务先执行
  const promises = items.map((item) =>
    queue.push(() => processItem(item), {
      priority: item < 5 ? 10 : 0,
    })
  );

  // 监控队列状态
  const monitor = setInterval(() => {
    console.log(queue.status);
  }, 100);

  const results = await Promise.all(promises);
  clearInterval(monitor);

  console.log('Final stats:', queue.status);
  console.log('Results:', results);
}

main();

5.4 优先级队列

/**
 * 多级优先级队列
 * 高优先级队列优先消费，低优先级队列按权重消费
 */
class PriorityAsyncQueue {
  constructor(concurrency = 4) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queues = new Map(); // priority -> task[]
    this.initPriorities([10, 5, 1]); // 高、中、低
  }

  initPriorities(levels) {
    for (const p of levels) {
      this.queues.set(p, []);
    }
  }

  push(task, priority = 5) {
    if (!this.queues.has(priority)) {
      this.queues.set(priority, []);
    }

    return new Promise((resolve, reject) => {
      this.queues.get(priority).push({ task, resolve, reject });
      this.run();
    });
  }

  getNext() {
    // 按优先级降序获取
    const priorities = [...this.queues.keys()].sort((a, b) => b - a);
    for (const p of priorities) {
      const queue = this.queues.get(p);
      if (queue.length > 0) {
        return queue.shift();
      }
    }
    return null;
  }

  run() {
    while (this.running < this.concurrency) {
      const item = this.getNext();
      if (!item) break;

      this.running++;
      Promise.resolve()
        .then(() => item.task())
        .then(item.resolve, item.reject)
        .finally(() => {
          this.running--;
          this.run();
        });
    }
  }

  get status() {
    const queues = {};
    for (const [p, q] of this.queues) {
      queues[p] = q.length;
    }
    return { running: this.running, queues };
  }
}

// 使用示例
const pq = new PriorityAsyncQueue(2);

// 低优先级任务
for (let i = 0; i < 20; i++) {
  pq.push(() => fetch(`/low/${i}`), 1);
}

// 高优先级任务（会插队）
for (let i = 0; i < 5; i++) {
  pq.push(() => fetch(`/high/${i}`), 10);
}

5.5 可取消任务队列

/**
 * 支持任务取消的异步队列
 * 取消时正在执行的任务会收到 AbortSignal
 */
class CancellableAsyncQueue {
  constructor(concurrency = 4) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = new Map(); // id -> { task, resolve, reject, controller }
    this.nextId = 0;
  }

  push(task, options = {}) {
    const id = ++this.nextId;
    const controller = new AbortController();

    const promise = new Promise((resolve, reject) => {
      this.queue.set(id, { task, resolve, reject, controller });
      this.run();
    });

    return {
      id,
      promise,
      cancel: () => this.cancel(id),
      controller,
    };
  }

  cancel(id) {
    const item = this.queue.get(id);
    if (item) {
      item.controller.abort();
      this.queue.delete(id);
      item.reject(new DOMException('Task cancelled', 'AbortError'));
      return true;
    }
    return false;
  }

  cancelAll() {
    const ids = [...this.queue.keys()];
    for (const id of ids) {
      this.cancel(id);
    }
  }

  run() {
    while (this.running < this.concurrency && this.queue.size > 0) {
      const [id, item] = this.queue.entries().next().value;
      this.queue.delete(id);
      this.running++;

      Promise.resolve()
        .then(() => item.task(item.controller.signal))
        .then(item.resolve, item.reject)
        .finally(() => {
          this.running--;
          this.run();
        });
    }
  }
}

// 使用示例
const queue = new CancellableAsyncQueue(2);

// 支持取消的 fetch
async function cancellableFetch(url, signal) {
  const response = await fetch(url, { signal });
  return response.json();
}

const tasks = urls.map((url) => queue.push((signal) => cancellableFetch(url, signal)));

// 5 秒后取消所有任务
setTimeout(() => {
  queue.cancelAll();
}, 5000);

try {
  const results = await Promise.all(tasks.map((t) => t.promise));
} catch (e) {
  if (e.name === 'AbortError') {
    console.log('任务被取消');
  }
}

5.6 批处理模式

/**
 * 分批执行：将任务分成固定大小的批次，批次内并发，批次间串行
 * @param {Array} items - 任务数组
 * @param {number} batchSize - 批次大小
 * @param {Function} fn - 处理函数
 * @returns {Promise<Array>} 结果数组
 */
async function batchProcess(items, batchSize, fn) {
  const results = [];
  for (let i = 0; i < items.length; i += batchSize) {
    const batch = items.slice(i, i + batchSize);
    const batchResults = await Promise.all(batch.map(fn));
    results.push(...batchResults);
    console.log(`完成批次 ${Math.floor(i / batchSize) + 1}`);
  }
  return results;
}

// 使用示例
const ids = Array.from({ length: 100 }, (_, i) => i);
const results = await batchProcess(ids, 10, async (id) => {
  const res = await fetch(`/api/user/${id}`);
  return res.json();
});

5.7 滑动窗口模式

/**
 * 滑动窗口并发：维护固定数量的 worker，每个 worker 从共享索引拉取任务
 * 相比 p-limit，避免入队 N 个 Promise 的内存开销
 */
async function slidingWindow(items, concurrency, fn) {
  const results = new Array(items.length);
  let nextIndex = 0;

  async function worker() {
    while (nextIndex < items.length) {
      const index = nextIndex++;
      try {
        results[index] = await fn(items[index], index);
      } catch (error) {
        results[index] = { error };
      }
    }
  }

  const workers = Array.from({ length: concurrency }, () => worker());
  await Promise.all(workers);
  return results;
}

// 使用示例
const urls = Array.from({ length: 1000 }, (_, i) => `/api/${i}`);
const results = await slidingWindow(urls, 10, async (url) => {
  const res = await fetch(url);
  return res.json();
});

5.8 速率限制器

/**
 * 令牌桶限速器
 * 按固定速率发放令牌，任务必须拿到令牌才能执行
 */
class TokenBucketRateLimiter {
  constructor(rate, capacity) {
    this.rate = rate; // 令牌生成速率（个/秒）
    this.capacity = capacity || rate; // 桶容量
    this.tokens = capacity;
    this.lastRefillTime = Date.now();
    this.queue = [];
  }

  async acquire() {
    if (this._tryAcquire()) {
      return;
    }

    return new Promise((resolve) => {
      this.queue.push(resolve);
      this._scheduleRefill();
    });
  }

  _tryAcquire() {
    this._refill();
    if (this.tokens >= 1) {
      this.tokens -= 1;
      return true;
    }
    return false;
  }

  _refill() {
    const now = Date.now();
    const elapsed = (now - this.lastRefillTime) / 1000;
    this.tokens = Math.min(this.capacity, this.tokens + elapsed * this.rate);
    this.lastRefillTime = now;
  }

  _scheduleRefill() {
    const waitTime = (1 / this.rate) * 1000;
    setTimeout(() => {
      this._drainQueue();
    }, waitTime);
  }

  _drainQueue() {
    while (this.queue.length > 0 && this._tryAcquire()) {
      const resolve = this.queue.shift();
      resolve();
    }
    if (this.queue.length > 0) {
      this._scheduleRefill();
    }
  }

  get status() {
    return {
      tokens: this.tokens,
      capacity: this.capacity,
      queueLength: this.queue.length,
    };
  }
}

// 使用示例：每秒最多 5 个请求
const limiter = new TokenBucketRateLimiter(5, 5);

async function rateLimitedFetch(url) {
  await limiter.acquire();
  return fetch(url);
}

5.9 组合：并发 + 限速

/**
 * 组合并发控制与速率限制
 */
class ConcurrencyRateLimiter {
  constructor({ concurrency, rate, capacity }) {
    this.queue = new AsyncQueue(concurrency);
    this.limiter = new TokenBucketRateLimiter(rate, capacity);
  }

  async push(task) {
    return this.queue.push(async () => {
      await this.limiter.acquire();
      return task();
    });
  }
}

// 使用：最多 3 个并发，且每秒最多 5 个请求
const combined = new ConcurrencyRateLimiter({
  concurrency: 3,
  rate: 5,
  capacity: 5,
});

for (const url of urls) {
  combined.push(() => fetch(url));
}

6. 对比分析

6.1 并发控制方案对比

6.2 p-limit vs AsyncQueue

6.3 FIFO vs LIFO vs 优先级

6.4 浏览器 vs Node.js 并发差异

6.5 与其他语言的对比

JavaScript 的并发控制与 Python asyncio 最相似，都基于事件循环 + Promise/Future 模型。

7. 常见陷阱

7.1 陷阱：Promise.all 不限制并发

// 错误：以为 Promise.all 会自动限制
const results = await Promise.all(urls.map(fetch));
// 实际：所有 fetch 同时启动，可能触发 429

正确做法：

const limit = pLimit(6);
const results = await Promise.all(
  urls.map((url) => limit(() => fetch(url)))
);

7.2 陷阱：map 立即创建所有 Promise

// 错误：limit() 返回的 Promise 在 map 时就已创建
const promises = urls.map((url) => limit(() => fetch(url)));
// 此时所有任务都已入队
await Promise.all(promises);

如果 urls 有 10 万个，会立即在内存中创建 10 万个 Promise 对象。对超大数据集应使用滑动窗口：

await slidingWindow(urls, 6, fetch);

7.3 陷阱：finally 中抛错

// 错误：finally 中抛错会覆盖 try 的返回值
const run = async (fn, resolve, reject) => {
  try {
    const result = await fn();
    resolve(result);
  } catch (e) {
    reject(e);
  } finally {
    cleanup(); // 如果 cleanup 抛错，会改变 Promise 状态
  }
};

正确做法：

const run = async (fn, resolve, reject) => {
  try {
    const result = await fn();
    resolve(result);
  } catch (e) {
    reject(e);
  } finally {
    try {
      await cleanup();
    } catch (cleanupError) {
      console.error('cleanup failed:', cleanupError);
    }
  }
};

7.4 陷阱：闭包捕获共享变量

// 错误：i 是共享变量，所有任务可能拿到相同的 i
for (var i = 0; i < urls.length; i++) {
  limit(() => fetch(urls[i])); // i 是同一个变量
}

正确做法（使用 let）：

for (let i = 0; i < urls.length; i++) {
  limit(() => fetch(urls[i]));
}

或使用 forEach/map：

urls.forEach((url) => limit(() => fetch(url)));

7.5 陷阱：任务异常导致整批失败

// 错误：一个任务失败，Promise.all 立即 reject
const results = await Promise.all(
  urls.map((url) => limit(() => fetch(url)))
);

改进：使用 Promise.allSettled：

const settled = await Promise.allSettled(
  urls.map((url) => limit(() => fetch(url)))
);
const successful = settled.filter((s) => s.status === 'fulfilled').map((s) => s.value);
const failed = settled.filter((s) => s.status === 'rejected').map((s) => s.reason);

7.6 陷阱：未处理 unhandledrejection

// 错误：limit() 返回的 Promise 若未 catch，reject 会成为 unhandledrejection
urls.map((url) => limit(() => fetch(url)));
// 如果 fetch 失败，且没有 await Promise.all，会触发 unhandledrejection

正确做法：确保所有 Promise 都被 await 或 catch：

const promises = urls.map((url) =>
  limit(() => fetch(url)).catch((e) => ({ error: e, url }))
);
const results = await Promise.all(promises);

7.7 陷阱：并发度设置过高

// 错误：并发度等于任务数，等于无限制
const limit = pLimit(urls.length);
await Promise.all(urls.map((url) => limit(() => fetch(url))));

建议：

• 浏览器：4-6（HTTP/1.1）或 20-50（HTTP/2）
• Node.js API：10-100
• 数据库：连接池大小（通常 10-30）

7.8 陷阱：任务函数有副作用

// 错误：任务函数依赖外部状态，执行顺序影响结果
let counter = 0;
const limit = pLimit(3);
const results = await Promise.all(
  items.map((item) =>
    limit(() => {
      counter++; // 共享状态，并发执行时不可预测
      return processItem(item, counter);
    })
  )
);

改进：避免共享状态，或将状态作为参数传入：

const results = await Promise.all(
  items.map((item, index) =>
    limit(() => processItem(item, index)) // index 是确定的
  )
);

7.9 陷阱：动态修改并发度

// p-limit 的 concurrency 在创建时固定，无法动态调整
const limit = pLimit(3);
// 后续无法改为 5

解决：使用可变并发度的实现：

class MutableConcurrencyQueue {
  constructor(initial = 4) {
    this.concurrency = initial;
    this.running = 0;
    this.queue = [];
  }

  setConcurrency(n) {
    this.concurrency = n;
    this.run(); // 立即尝试启动更多任务
  }

  // ... 其他方法
}

7.10 陷阱：内存泄漏

// 错误：队列中累积大量任务，长时间未消费
const queue = new AsyncQueue(2);
setInterval(() => {
  queue.push(() => fetchData()); // 持续入队，但消费速度跟不上
}, 100);
// 队列可能无限增长，最终 OOM

解决：设置队列上限：

class BoundedAsyncQueue extends AsyncQueue {
  constructor(concurrency, maxSize = 1000) {
    super(concurrency);
    this.maxSize = maxSize;
  }

  push(task) {
    if (this.queue.length >= this.maxSize) {
      return Promise.reject(new Error('Queue is full'));
    }
    return super.push(task);
  }
}

8. 工程实践

8.1 动态并发度调整

根据网络条件动态调整并发度：

class AdaptiveQueue extends AsyncQueue {
  constructor(initial = 4) {
    super(initial);
    this.successWindow = [];
    this.failureWindow = [];
    this.windowSize = 20;
  }

  _recordSuccess(duration) {
    this.successWindow.push(duration);
    if (this.successWindow.length > this.windowSize) {
      this.successWindow.shift();
    }
    this._adjustConcurrency();
  }

  _recordFailure() {
    this.failureWindow.push(Date.now());
    if (this.failureWindow.length > this.windowSize) {
      this.failureWindow.shift();
    }
    this._adjustConcurrency();
  }

  _adjustConcurrency() {
    const failureRate = this.failureWindow.length / this.windowSize;
    const avgDuration = this.successWindow.length > 0
      ? this.successWindow.reduce((a, b) => a + b, 0) / this.successWindow.length
      : 0;

    if (failureRate > 0.3) {
      // 失败率高，降低并发
      this.concurrency = Math.max(1, Math.floor(this.concurrency * 0.7));
      console.log(`降低并发度到 ${this.concurrency}`);
    } else if (failureRate < 0.05 && avgDuration < 500) {
      // 成功率高且响应快，提升并发
      this.concurrency = Math.min(20, this.concurrency + 1);
      console.log(`提升并发度到 ${this.concurrency}`);
    }
  }
}

8.2 断路器集成

/**
 * 断路器：连续失败时熔断，保护下游服务
 */
class CircuitBreaker {
  constructor(options = {}) {
    this.failureThreshold = options.failureThreshold || 5;
    this.resetTimeout = options.resetTimeout || 30000;
    this.failures = 0;
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
    this.openedAt = 0;
  }

  async exec(fn) {
    if (this.state === 'OPEN') {
      if (Date.now() - this.openedAt > this.resetTimeout) {
        this.state = 'HALF_OPEN';
      } else {
        throw new Error('Circuit breaker is open');
      }
    }

    try {
      const result = await fn();
      this._onSuccess();
      return result;
    } catch (e) {
      this._onFailure();
      throw e;
    }
  }

  _onSuccess() {
    this.failures = 0;
    this.state = 'CLOSED';
  }

  _onFailure() {
    this.failures++;
    if (this.failures >= this.failureThreshold) {
      this.state = 'OPEN';
      this.openedAt = Date.now();
    }
  }
}

// 组合使用
const queue = new AsyncQueue(5);
const breaker = new CircuitBreaker();

async function safeFetch(url) {
  return queue.push(() =>
    breaker.exec(() => fetch(url).then((r) => {
      if (!r.ok) throw new Error(`HTTP ${r.status}`);
      return r.json();
    }))
  );
}

8.3 重试机制

/**
 * 带指数退避的重试
 */
async function withRetry(fn, options = {}) {
  const maxRetries = options.maxRetries || 3;
  const baseDelay = options.baseDelay || 1000;
  const maxDelay = options.maxDelay || 30000;

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn();
    } catch (error) {
      if (attempt === maxRetries) {
        throw error;
      }

      // 指数退避 + 抖动
      const delay = Math.min(
        maxDelay,
        baseDelay * Math.pow(2, attempt) + Math.random() * 1000
      );
      console.log(`第 ${attempt + 1} 次重试，等待 ${delay}ms`);
      await new Promise((r) => setTimeout(r, delay));
    }
  }
}

// 使用
const results = await Promise.all(
  urls.map((url) =>
    limit(() => withRetry(() => fetch(url), { maxRetries: 3 }))
  )
);

8.4 可观测性：日志与指标

class ObservableQueue extends AsyncQueue {
  constructor(options = {}) {
    super(options);
    this.metrics = {
      totalTasks: 0,
      totalSuccess: 0,
      totalFailure: 0,
      totalWaitTime: 0,
      totalRunTime: 0,
      maxWaitTime: 0,
      maxRunTime: 0,
    };
  }

  push(task, options = {}) {
    const enqueuedAt = Date.now();
    this.metrics.totalTasks++;

    return super.push(async () => {
      const waitTime = Date.now() - enqueuedAt;
      this.metrics.totalWaitTime += waitTime;
      this.metrics.maxWaitTime = Math.max(this.metrics.maxWaitTime, waitTime);

      const startAt = Date.now();
      try {
        const result = await task();
        this.metrics.totalSuccess++;
        const runTime = Date.now() - startAt;
        this.metrics.totalRunTime += runTime;
        this.metrics.maxRunTime = Math.max(this.metrics.maxRunTime, runTime);
        return result;
      } catch (e) {
        this.metrics.totalFailure++;
        throw e;
      }
    }, options);
  }

  getMetrics() {
    const total = this.metrics.totalSuccess + this.metrics.totalFailure;
    return {
      ...this.metrics,
      successRate: total > 0 ? this.metrics.totalSuccess / total : 0,
      avgWaitTime: total > 0 ? this.metrics.totalWaitTime / total : 0,
      avgRunTime: total > 0 ? this.metrics.totalRunTime / total : 0,
      throughput: total > 0 ? total / (this.metrics.totalRunTime / 1000) : 0,
    };
  }
}

8.5 React 中的并发控制

import { useRef, useCallback } from 'react';

function useConcurrencyLimiter(concurrency = 4) {
  const queueRef = useRef(null);

  if (!queueRef.current) {
    queueRef.current = new AsyncQueue(concurrency);
  }

  const push = useCallback((task) => {
    return queueRef.current.push(task);
  }, []);

  const status = queueRef.current.status;

  return { push, status };
}

// 使用示例
function FileUploader({ files }) {
  const { push, status } = useConcurrencyLimiter(3);

  const handleUpload = async () => {
    const results = await Promise.all(
      files.map((file) =>
        push(() => uploadFile(file))
      )
    );
    console.log(results);
  };

  return (
    <div>
      <button onClick={handleUpload}>上传 {files.length} 个文件</button>
      <p>运行中: {status.running}，等待中: {status.pending}</p>
    </div>
  );
}

8.6 Node.js 中的流式并发

const { Readable } = require('stream');

/**
 * 流式并发处理：边读取边处理，避免一次性加载所有数据
 */
async function processStream(inputStream, concurrency, processor) {
  const queue = new AsyncQueue(concurrency);
  const outputQueue = [];
  let inputEnded = false;

  await new Promise((resolve, reject) => {
    inputStream.on('data', (chunk) => {
      const promise = queue.push(() => processor(chunk));
      outputQueue.push(promise);
    });

    inputStream.on('end', () => {
      inputEnded = true;
      Promise.all(outputQueue).then(resolve, reject);
    });

    inputStream.on('error', reject);
  });
}

// 使用示例：处理大文件
const fs = require('fs');
const readline = require('readline');

const rl = readline.createInterface({
  input: fs.createReadStream('large-file.txt'),
});

const lines = [];
rl.on('line', (line) => lines.push(line));
rl.on('close', async () => {
  await processStream(lines, 10, async (line) => {
    // 处理每行
    return processLine(line);
  });
});

8.7 性能基准测试

async function benchmark() {
  const sizes = [100, 1000, 10000, 100000];
  const concurrencies = [1, 4, 8, 16, 32];

  for (const size of sizes) {
    for (const concurrency of concurrencies) {
      const tasks = Array.from({ length: size }, (_, i) => async () => {
        await new Promise((r) => setTimeout(r, 10));
        return i;
      });

      const start = performance.now();
      const limit = pLimit(concurrency);
      await Promise.all(tasks.map((t) => limit(t)));
      const elapsed = performance.now() - start;

      console.log(
        `size=${size}, concurrency=${concurrency}: ${elapsed.toFixed(0)}ms`
      );
    }
  }
}

benchmark();
// 典型输出（Node.js 18, M1 Pro）：
// size=100, concurrency=1: 1005ms
// size=100, concurrency=4: 260ms
// size=100, concurrency=8: 140ms
// size=100, concurrency=16: 90ms
// size=1000, concurrency=16: 640ms
// size=10000, concurrency=16: 6300ms

8.8 内存优化

对超大队列使用生成器而非数组：

/**
 * 生成器驱动的并发控制
 * 适用于任务来源是流式或生成器的情况
 */
async function processGenerator(taskGenerator, concurrency, fn) {
  const workers = Array.from({ length: concurrency }, async () => {
    for (const task of taskGenerator) {
      await fn(task);
    }
  });

  await Promise.all(workers);
}

// 使用示例：处理大文件行
async function* lineGenerator(filePath) {
  const rl = readline.createInterface({
    input: fs.createReadStream(filePath),
  });
  for await (const line of rl) {
    yield line;
  }
}

await processGenerator(lineGenerator('huge.log'), 8, async (line) => {
  // 处理每行
});

9. 案例研究

9.1 案例一：图片批量上传组件

需求：用户选择 100 张图片上传到 CDN，要求显示进度，支持暂停/继续/取消。

class ImageUploader {
  constructor(options = {}) {
    this.queue = new AsyncQueue({ concurrency: options.concurrency || 3 });
    this.uploads = new Map(); // id -> { progress, status, controller }
    this.listeners = new Set();
  }

  on(listener) {
    this.listeners.add(listener);
    return () => this.listeners.delete(listener);
  }

  _emit(event) {
    for (const listener of this.listeners) {
      listener(event);
    }
  }

  async upload(file) {
    const id = Date.now() + Math.random();
    const uploadState = {
      id,
      file,
      progress: 0,
      status: 'pending',
      controller: new AbortController(),
    };
    this.uploads.set(id, uploadState);
    this._emit({ type: 'added', id });

    try {
      const result = await this.queue.push(async () => {
        uploadState.status = 'uploading';
        this._emit({ type: 'started', id });

        const formData = new FormData();
        formData.append('file', file);

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

        // 监听上传进度（使用 XMLHttpRequest 更易实现）
        return await this._uploadWithProgress(file, uploadState);
      });

      uploadState.status = 'completed';
      uploadState.progress = 100;
      this._emit({ type: 'completed', id, result });
      return result;
    } catch (error) {
      if (error.name === 'AbortError') {
        uploadState.status = 'cancelled';
        this._emit({ type: 'cancelled', id });
      } else {
        uploadState.status = 'failed';
        uploadState.error = error;
        this._emit({ type: 'failed', id, error });
      }
      throw error;
    }
  }

  _uploadWithProgress(file, state) {
    return new Promise((resolve, reject) => {
      const xhr = new XMLHttpRequest();
      xhr.open('POST', '/api/upload');
      xhr.upload.onprogress = (e) => {
        if (e.lengthComputable) {
          state.progress = Math.round((e.loaded / e.total) * 100);
          this._emit({ type: 'progress', id: state.id, progress: state.progress });
        }
      };
      xhr.onload = () => {
        if (xhr.status === 200) {
          resolve(JSON.parse(xhr.responseText));
        } else {
          reject(new Error(`Upload failed: ${xhr.status}`));
        }
      };
      xhr.onerror = () => reject(new Error('Network error'));
      state.controller.signal.addEventListener('abort', () => {
        xhr.abort();
        reject(new DOMException('Aborted', 'AbortError'));
      });

      const formData = new FormData();
      formData.append('file', file);
      xhr.send(formData);
    });
  }

  cancel(id) {
    const state = this.uploads.get(id);
    if (state && (state.status === 'pending' || state.status === 'uploading')) {
      state.controller.abort();
    }
  }

  cancelAll() {
    for (const state of this.uploads.values()) {
      if (state.status === 'pending' || state.status === 'uploading') {
        state.controller.abort();
      }
    }
  }

  pauseAll() {
    this.queue.pause();
  }

  resumeAll() {
    this.queue.resume();
  }

  get summary() {
    const states = [...this.uploads.values()];
    return {
      total: states.length,
      pending: states.filter((s) => s.status === 'pending').length,
      uploading: states.filter((s) => s.status === 'uploading').length,
      completed: states.filter((s) => s.status === 'completed').length,
      failed: states.filter((s) => s.status === 'failed').length,
      cancelled: states.filter((s) => s.status === 'cancelled').length,
      totalProgress:
        states.reduce((sum, s) => sum + s.progress, 0) / states.length,
    };
  }
}

// React 组件使用
function UploaderComponent() {
  const [uploader] = useState(() => new ImageUploader({ concurrency: 3 }));
  const [summary, setSummary] = useState(uploader.summary);

  useEffect(() => {
    return uploader.on((event) => {
      setSummary({ ...uploader.summary });
    });
  }, [uploader]);

  const handleFiles = async (files) => {
    await Promise.all(
      Array.from(files).map((file) =>
        uploader.upload(file).catch((e) => console.error(e))
      )
    );
  };

  return (
    <div>
      <input
        type="file"
        multiple
        onChange={(e) => handleFiles(e.target.files)}
      />
      <progress value={summary.totalProgress} max="100" />
      <p>
        完成 {summary.completed}/{summary.total}，
        失败 {summary.failed}
      </p>
      <button onClick={() => uploader.pauseAll()}>暂停</button>
      <button onClick={() => uploader.resumeAll()}>继续</button>
      <button onClick={() => uploader.cancelAll()}>取消全部</button>
    </div>
  );
}