全栈攻城狮

Appearance

异步编程进阶 ​

会用 Promise 与 async/await 只是起点。真实项目里，请求要限制并发、失败要重试、超时要取消、连续点击要防抖、大列表要流式处理。这些"控制异步"的能力，是进阶开发者与初学者的分水岭。本文用可运行的代码，逐个攻破异步控制的核心模式。

一、异步控制的三类问题 ​

问题表现后果
并发失控一次性发出数百个请求网络拥塞、服务器压力过大
竞态多个异步结果乱序到达旧结果覆盖新结果、状态错乱
背压生产速度远超消费速度内存堆积、队列无限膨胀
javascript
// 并发失控示例：循环里直接发起请求，无任何限制
for (let i = 0; i < 1000; i++) {
  fetch(`/api/item/${i}`); // 1000 个并发请求同时发出
}

应对思路分别对应：限流器（管并发）、竞态防护（管顺序）、流式消费（管速度）。

二、并发限制实现 ​

2.1 手动并发池 ​

javascript
async function runPool(tasks, limit) {
  const results = new Array(tasks.length);
  let index = 0;

  async function worker() {
    while (index < tasks.length) {
      const i = index++; // 原子地取任务
      results[i] = await tasks[i](); // 执行并把结果放回原位
    }
  }

  const workers = Array.from({ length: Math.min(limit, tasks.length) }, worker);
  await Promise.all(workers);
  return results;
}

// 用法
const tasks = Array.from({ length: 10 }, (_, i) => () =>
  new Promise((res) => setTimeout(() => res(i), 100))
);
runPool(tasks, 3).then((r) => console.log(r)); // 最多 3 个并发，耗时约为 4 组 × 100ms

2.2 通用限流器（p-limit 原理） ​

p-limit 的核心是信号量 + 等待队列：

javascript
function pLimit(limit) {
  const queue = [];   // 等待执行的任务
  let active = 0;     // 当前执行中的数量

  function next() {
    active--;
    if (queue.length) {
      queue.shift()(); // 有等待者则唤醒一个
    }
  }

  return async function enqueue(fn, ...args) {
    await new Promise((resolve) => queue.push(resolve)); // 排队等待名额
    active++;
    try {
      return await fn(...args);
    } finally {
      next();
    }
  };
}

const limit = pLimit(2);
const jobs = [1, 2, 3, 4].map((n) => limit(() => fetch(`/api/${n}`)));
await Promise.all(jobs);
实现特点
并发池（Worker 模式）预先创建固定数量 worker，任务从共享队列取
信号量限流（p-limit）每个任务先排队拿"名额"，拿不到就等待
时间窗口限流限制单位时间内执行次数

三、Promise 扩展模式 ​

3.1 重试（带退避） ​

javascript
async function withRetry(fn, { retries = 3, delay = 300 } = {}) {
  let lastError;
  for (let attempt = 0; attempt <= retries; attempt++) {
    try {
      return await fn();
    } catch (err) {
      lastError = err;
      if (attempt === retries) break;
      await sleep(delay * Math.pow(2, attempt)); // 指数退避
    }
  }
  throw lastError;
}
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// 网络抖动时自动重试 3 次
const data = await withRetry(() => fetch("/api/data").then((r) => r.json()));

3.2 超时 ​

javascript
function withTimeout(promise, ms, message = "操作超时") {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error(message)), ms);
    promise.then(
      (v) => { clearTimeout(timer); resolve(v); },
      (e) => { clearTimeout(timer); reject(e); }
    );
  });
}

await withTimeout(fetch("/slow"), 5000); // 5 秒内不返回则抛错

3.3 取消 ​

javascript
function cancellable(fn) {
  let cancelled = false;
  const promise = new Promise((resolve, reject) => {
    fn(resolve, reject, () => cancelled);
  });
  return {
    promise,
    cancel: () => { cancelled = true; },
  };
}

四、AbortController 深入 ​

AbortController 是标准的取消机制，核心是两个成员：

成员作用
controller.signal传给异步 API 的取消信号
controller.abort()触发取消，signal 变为 aborted

4.1 取消 fetch ​

javascript
const controller = new AbortController();

fetch("/api/large", { signal: controller.signal }).catch((err) => {
  if (err.name === "AbortError") console.log("请求已取消");
});

// 5 秒后取消请求
setTimeout(() => controller.abort(), 5000);

4.2 跨平台取消信号 ​

同一个 signal 可同时传给 fetch、事件监听、流读取，实现"一处取消，处处生效"：

javascript
const controller = new AbortController();
const { signal } = controller;

fetch("/api/data", { signal });
document.addEventListener("click", handler, { signal }); // 取消时自动移除监听
readableStream.pipeTo(writable, { signal });             // 取消时中止流

// 监听取消事件，做清理工作
signal.addEventListener("abort", () => {
  console.log("收到取消信号，执行清理");
});
场景传统做法AbortController
fetch 取消无标准方案signal 支持
移除监听器手动 removeEventListener{ signal } 自动移除
流中止手动 cancel(){ signal } 自动中止
多 API 联动各自单独处理共享一个 signal

五、可取消 Promise 设计 ​

让"取消"真正停掉底层任务，需要任务本身配合 signal：

javascript
function cancelableTask(signal) {
  return new Promise((resolve, reject) => {
    if (signal.aborted) return reject(new DOMException("取消", "AbortError"));

    const timer = setTimeout(() => {
      console.log("任务执行完成");
      resolve("done");
    }, 2000);

    signal.addEventListener("abort", () => {
      clearTimeout(timer); // 真正停止底层计时器
      reject(new DOMException("取消", "AbortError"));
    }, { once: true });
  });
}

const controller = new AbortController();
const task = cancelableTask(controller.signal).catch((e) => console.log(e.name));
setTimeout(() => controller.abort(), 500); // 500ms 后取消

要点：取消信号要下穿到底层资源，只 reject 外层 Promise 而底层继续运行，只是"假装取消"。

六、异步任务队列 ​

6.1 串行队列 ​

javascript
class SerialQueue {
  constructor() { this.running = Promise.resolve(); }
  add(task) {
    this.running = this.running.then(() => task());
    return this.running; // 返回当前任务自己的 Promise
  }
}
const queue = new SerialQueue();
[1, 2, 3].forEach((n) =>
  queue.add(() => new Promise((r) => setTimeout(() => { console.log(n); r(); }, 100)))
); // 依次输出 1 2 3，完全串行

6.2 并行队列（带限流） ​

javascript
class ParallelQueue {
  constructor(limit) {
    this.limit = limit;
    this.active = 0;
    this.waiting = [];
  }
  add(fn) {
    return new Promise((resolve, reject) => {
      this.waiting.push({ fn, resolve, reject });
      this.pump();
    });
  }
  pump() {
    while (this.active < this.limit && this.waiting.length) {
      const { fn, resolve, reject } = this.waiting.shift();
      this.active++;
      fn().then(resolve, reject).finally(() => {
        this.active--;
        this.pump();
      });
    }
  }
}

6.3 优先级队列 ​

javascript
class PriorityQueue {
  constructor() { this.tasks = []; }
  add(priority, fn) {
    // 按优先级降序插入（数字越大越先执行）
    const index = this.tasks.findIndex((t) => t.priority < priority);
    if (index === -1) this.tasks.push({ priority, fn });
    else this.tasks.splice(index, 0, { priority, fn });
    return this;
  }
  async run() {
    while (this.tasks.length) {
      const { fn } = this.tasks.shift();
      await fn();
    }
  }
}
const q = new PriorityQueue();
q.add(1, async () => console.log("低优先级"));
q.add(3, async () => console.log("高优先级"));
q.add(2, async () => console.log("中优先级"));
await q.run(); // 高优先级 → 中优先级 → 低优先级
队列类型适用场景
串行队列写文件、事务类操作，必须按顺序
并行队列（限流）批量下载、图片处理
优先级队列任务有轻重缓急（如渲染帧任务）

七、竞态条件处理 ​

7.1 请求竞态：只认最新 ​

javascript
function latestRequest(fn) {
  let seq = 0;
  return async (...args) => {
    const mySeq = ++seq; // 每次调用拿到递增序号
    const result = await fn(...args);
    if (mySeq !== seq) throw new Error("请求已过期"); // 已被更新的请求覆盖
    return result;
  };
}

const fetchLatest = latestRequest((kw) => fetch(`/api/search?q=${kw}`).then((r) => r.json()));

fetchLatest("苹果"); // 慢响应
fetchLatest("香蕉"); // 快响应，先返回
// “苹果”的响应到达时 mySeq !== seq → 判定过期，丢弃

7.2 防抖：合并连续触发 ​

javascript
function debounceLatest(fn, wait = 300) {
  let timer;
  return (...args) => {
    clearTimeout(timer);
    timer = setTimeout(() => fn(...args), wait);
  };
}
// 输入框防抖：停止输入 300ms 后才真正请求
searchInput.addEventListener("input", debounceLatest((e) => doSearch(e.target.value)));
方案策略适用
序号法（latest）丢弃过期响应结果必须对应最新参数
防抖（debounce）合并高频触发输入搜索、窗口 resize
节流（throttle）固定频率执行滚动事件、拖拽

八、信号量概念与实现 ​

信号量（Semaphore） 是控制并发资源的经典抽象：维护一个计数器，acquire 时计数减一（为 0 则等待），release 时计数加一（唤醒等待者）：

javascript
class Semaphore {
  constructor(limit) { this.limit = limit; this.active = 0; this.waiters = []; }
  async acquire() {
    if (this.active < this.limit) {
      this.active++;
      return;
    }
    await new Promise((resolve) => this.waiters.push(resolve));
  }
  release() {
    this.active--;
    const next = this.waiters.shift();
    if (next) {
      this.active++;
      next(); // 唤醒一个等待者并移交名额
    }
  }
}

const sem = new Semaphore(2);
async function guardedWork(id) {
  await sem.acquire();
  try {
    console.log(`任务 ${id} 开始`);
    await sleep(500);
    console.log(`任务 ${id} 结束`);
  } finally {
    sem.release(); // 无论成败都释放
  }
}
[1, 2, 3, 4].forEach(guardedWork); // 始终最多 2 个并发

release 必须放在 finally 中，防止任务抛错后名额被永久占用。

九、async 生成器与流式处理 ​

async function* 可以异步产出数据，配合 for await...of 实现边生产边消费，解决背压：

javascript
async function* streamNumbers(limit) {
  for (let i = 1; i <= limit; i++) {
    await sleep(100);           // 模拟异步生产
    yield i;                    // 产出一个，消费方立即拿到
  }
}

(async () => {
  let sum = 0;
  for await (const n of streamNumbers(5)) {
    sum += n;                   // 消费者按节奏处理，不积压
    console.log("收到", n);
  }
  console.log("总和", sum); // 15
})();
特性普通生成器async 生成器
产出同步值Promise 值
消费for...offor await...of
典型用途无限序列、懒计算分页拉取、大文件逐行处理
javascript
// 流式处理分页 API：一次只取一页，处理完再取下一页
async function* paginate(getPage, total) {
  for (let page = 1; page <= total; page++) {
    yield getPage(page);
  }
}

十、async/await 错误处理最佳实践 ​

实践说明
用 try/catch 包住业务代码await 抛出的异常必须被捕获，否则是未处理拒绝
顶层用 .catch模块顶层无法 try/catch，挂 .catch 兜底
finally 做清理释放资源、关闭连接、清除 loading 状态
区分"预期错误"与"程序错误"预期错误可返回默认值，程序错误应上抛
不要吞掉异常catch 后打印日志再决定是否继续
Promise 并发用 Promise.allSettled部分失败不影响整体结果收集
javascript
async function loadUser() {
  try {
    const res = await fetch("/api/user");
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return await res.json();
  } catch (err) {
    console.error("加载用户失败", err);
    return { fallback: true }; // 预期错误：给默认值
  } finally {
    loading = false; // 无论成败都清理
  }
}

// 并发任务：不因单个失败而中断整体
const results = await Promise.allSettled(
  urls.map((u) => fetch(u).then((r) => r.json()))
);
const ok = results.filter((r) => r.status === "fulfilled").map((r) => r.value);

把限流、重试、超时、取消、队列、竞态防护这些模式组装起来，就能应对绝大多数生产级异步场景。每个模式单独理解并不难，难的是组合时保持"取消信号下穿、资源 finally 释放、并发可控"这三条底线。