异步并发控制

1. 并发控制问题

1.1 无限制并发的风险

//  同时发起 1000 个请求
const urls = Array.from({ length: 1000 }, (_, i) => `/api/item/${i}`);
await Promise.all(urls.map((url) => fetch(url)));
// 可能导致：
// - 浏览器连接数耗尽（Chrome 同域最多 6 个）
// - 服务器拒绝服务（429 Too Many Requests）
// - 内存溢出

1.2 并发控制目标

• 限制同时执行的异步操作数量
• 保持最大并发度以提高吞吐量
• 任务完成后自动拉入下一个任务

2. p-limit 模式

2.1 基本实现

function pLimit(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));
      }
    });
  };

  return enqueue;
}

2.2 使用方式

const limit = pLimit(3); // 最多 3 个并发

const results = await Promise.all([
  limit(() => fetch('/api/1')),
  limit(() => fetch('/api/2')),
  limit(() => fetch('/api/3')),
  limit(() => fetch('/api/4')), // 等待前面的完成
  limit(() => fetch('/api/5')),
]);

2.3 动态并发度

const limit = pLimit(navigator.connection?.effectiveType === '4g' ? 6 : 3);

3. 异步队列实现

3.1 完整的任务队列

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

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

  run() {
    while (this.running < this.concurrency && this.queue.length > 0) {
      const { task, resolve, reject } = this.queue.shift();
      this.running++;

      task()
        .then(resolve)
        .catch(reject)
        .finally(() => {
          this.running--;
          this.run();
        });
    }
  }

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

  get pending() {
    return this.running;
  }
}

3.2 优先级队列

class PriorityAsyncQueue {
  constructor(concurrency = 4) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queues = {
      high: [],
      normal: [],
      low: [],
    };
  }

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

  getNext() {
    for (const priority of ['high', 'normal', 'low']) {
      if (this.queues[priority].length > 0) {
        return this.queues[priority].shift();
      }
    }
    return null;
  }

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

      this.running++;
      const { task, resolve, reject } = item;

      task()
        .then(resolve)
        .catch(reject)
        .finally(() => {
          this.running--;
          this.run();
        });
    }
  }
}

3.3 可取消队列

class CancellableAsyncQueue {
  constructor(concurrency = 4) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = new Map(); // id → task
    this.nextId = 0;
  }

  push(task) {
    const id = this.nextId++;
    return {
      id,
      promise: new Promise((resolve, reject) => {
        this.queue.set(id, { task, resolve, reject });
        this.run();
      }),
    };
  }

  cancel(id) {
    const item = this.queue.get(id);
    if (item) {
      this.queue.delete(id);
      item.reject(new Error('Cancelled'));
      return true;
    }
    return false;
  }

  cancelAll() {
    for (const [id, item] of this.queue) {
      item.reject(new Error('Cancelled'));
      this.queue.delete(id);
    }
  }

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

      task()
        .then(resolve)
        .catch(reject)
        .finally(() => {
          this.running--;
          this.run();
        });
    }
  }
}

4. 批处理模式

4.1 分批执行

async function batchProcess(items, batchSize, processFn) {
  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(processFn));
    results.push(...batchResults);
  }

  return results;
}

// 使用
const allResults = await batchProcess(urls, 10, (url) => fetch(url));

4.2 滑动窗口

async function slidingWindow(items, concurrency, processFn) {
  const results = new Array(items.length);
  let nextIndex = 0;

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

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

5. 实际应用

5.1 文件上传

class Uploader {
  constructor(maxConcurrent = 3) {
    this.queue = new AsyncQueue(maxConcurrent);
  }

  async upload(file) {
    return this.queue.push(async () => {
      const formData = new FormData();
      formData.append('file', file);
      const response = await fetch('/api/upload', {
        method: 'POST',
        body: formData,
      });
      return response.json();
    });
  }
}

5.2 爬虫限速

class RateLimitedFetcher {
  constructor(requestsPerSecond = 5) {
    this.interval = 1000 / requestsPerSecond;
    this.lastRequestTime = 0;
  }

  async fetch(url) {
    const now = Date.now();
    const elapsed = now - this.lastRequestTime;
    const delay = Math.max(0, this.interval - elapsed);

    if (delay > 0) {
      await new Promise((r) => setTimeout(r, delay));
    }

    this.lastRequestTime = Date.now();
    return fetch(url);
  }
}