JavaScript 异步编程与事件循环 / Asynchronous JavaScript and the Event Loop

📅 创建时间：2026-07-28 🏷️ 标签：#EventLoop #Promise #AsyncAwait #Microtask #Concurrency 📚 前置知识：[[01-javascript-runtime-and-language]]

---

📋 本章目标

• 理解事件循环的完整运行周期：调用栈、任务队列、微任务队列、渲染管线
• 掌握微任务与宏任务的精确执行顺序及其可视化模型
• 深入理解 Promise 的状态机、链式调用与六大静态方法
• 理解 async/await 的底层机制与常见误区
• 掌握 AbortController 实现可取消的异步操作
• 掌握异步迭代器、生成器与 Streams API 的使用场景
• 能够识别并修复常见的异步陷阱（floating promise、竞态条件等）

---

第1部分：事件循环深度解析

1.1 单线程的 JavaScript 为何不阻塞？

JavaScript 在单个执行线程上运行，但通过**事件循环（Event Loop）**协调 I/O、计时器和用户交互。关键在于：耗时操作由宿主环境（浏览器 / libuv）处理，完成后再将回调推入队列，由事件循环调度执行。

┌─────────────────────────────────────────────────────────────┐
│                    JavaScript 运行时全景                     │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────────┐    ┌──────────────────────────────┐      │
│  │  Call Stack   │    │      Web APIs / libuv       │      │
│  │  (调用栈)     │    │  setTimeout, fetch,         │      │
│  │              │    │  fs.readFile, setImmediate   │      │
│  │  ┌────────┐  │    │                              │      │
│  │  │  fn3()  │  │    │  注册回调 → 宿主处理 →      │      │
│  │  │  fn2()  │  │    │  完成后回调入队              │      │
│  │  │  fn1()  │  │    └──────────────────────────────┘      │
│  │  └────────┘  │              ↓                            │
│  └──────────────┘    ┌──────────────────────────────┐      │
│         │            │       Task Queues            │      │
│         ↓            │  ┌────────────────────────┐  │      │
│  ┌──────────────┐    │  │  Microtask Queue       │  │      │
│  │  Event Loop  │←───│  │  Promise.then,         │  │      │
│  │  (事件循环)   │    │  │  queueMicrotask,       │  │      │
│  │              │    │  │  MutationObserver       │  │      │
│  │  "有任务吗？" │    │  └────────────────────────┘  │      │
│  │  "栈空了吗？" │    │  ┌────────────────────────┐  │      │
│  │  "渲染吗？"   │    │  │  Macrotask Queue       │  │      │
│  └──────────────┘    │  │  setTimeout, setInterval,│  │      │
│                      │  │  I/O, UI events          │  │      │
│                      │  └────────────────────────┘  │      │
│                      └──────────────────────────────┘      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.2 事件循环的一次完整 Tick

┌─────────────────────────────────────────────────────────────┐
│               浏览器事件循环：一次完整 Tick                    │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 从 Macrotask Queue 取出一个任务执行                      │
│     │                                                       │
│     ▼                                                       │
│  2. 该任务执行完毕（Call Stack 变空）                         │
│     │                                                       │
│     ▼                                                       │
│  3. 清空 Microtask Queue（包括执行过程中新产生的微任务）      │
│     │  循环执行直到微任务队列彻底为空                         │
│     │                                                       │
│     ▼                                                       │
│  4. 判断是否需要渲染（requestAnimationFrame 回调在此执行）    │
│     │  通常 60fps → 约 16.6ms 一次                           │
│     │                                                       │
│     ▼                                                       │
│  5. 回到步骤 1，取下一个 Macrotask                            │
│                                                             │
│  关键规则：                                                  │
│  • 一个 macrotask → 清空全部 microtask → 可能渲染 → 下一个  │
│  • 微任务可以插队，在同一次 tick 内全部执行完毕               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.3 Node.js 事件循环：libuv 六阶段

Node.js 的事件循环基于 libuv，分为六个阶段，每个阶段维护自己的 FIFO 回调队列。

┌─────────────────────────────────────────────────────────────┐
│              Node.js 事件循环：libuv 六阶段                   │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│    ┌──────────────────────────────────────────────┐        │
│    │                 timers                        │        │
│    │  setTimeout / setInterval 到期回调             │        │
│    └─────────────┬────────────────────────────────┘        │
│                  ▼                                          │
│    ┌──────────────────────────────────────────────┐        │
│    │           pending callbacks                   │        │
│    │  系统操作（如 TCP 错误）推迟到此处执行           │        │
│    └─────────────┬────────────────────────────────┘        │
│                  ▼                                          │
│    ┌──────────────────────────────────────────────┐        │
│    │           idle, prepare                       │        │
│    │  libuv 内部使用，一般无需关心                   │        │
│    └─────────────┬────────────────────────────────┘        │
│                  ▼                                          │
│    ┌──────────────────────────────────────────────┐        │
│    │               poll                            │        │
│    │  获取新的 I/O 事件；阻塞等待（若无过期 timer）  │        │
│    │  fs.readFile、网络请求等回调在此入队            │        │
│    └─────────────┬────────────────────────────────┘        │
│                  ▼                                          │
│    ┌──────────────────────────────────────────────┐        │
│    │               check                           │        │
│    │  setImmediate 回调在此执行                     │        │
│    └─────────────┬────────────────────────────────┘        │
│                  ▼                                          │
│    ┌──────────────────────────────────────────────┐        │
│    │           close callbacks                     │        │
│    │  socket.on('close', ...) 等清理回调            │        │
│    └─────────────┬────────────────────────────────┘        │
│                  │                                          │
│                  ▼ 下一轮循环（或退出）                      │
│                                                             │
│  两个特殊的"插队"队列（每个阶段切换时清空）：                 │
│  ┌──────────────────────────────────────────────┐        │
│  │  • nextTick Queue: process.nextTick()        │        │
│  │  • Microtask Queue: Promise.then,            │        │
│  │    queueMicrotask                             │        │
│  │  优先级：nextTick > Microtask                 │        │
│  └──────────────────────────────────────────────┘        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

第2部分：微任务与宏任务

2.1 两类队列的本质区别

┌─────────────────────────────────────────────────────────────┐
│                    Microtask vs Macrotask                     │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Macrotask（宏任务 / Task）                                  │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  来源：setTimeout, setInterval, I/O, UI 事件,       │   │
│  │        setImmediate(Node), MessageChannel           │   │
│  │  调度：每个事件循环 tick 只取一个执行                 │   │
│  │  特点：两次 macrotask 之间可能发生渲染               │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  Microtask（微任务 / Job）                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  来源：Promise.then/catch/finally, queueMicrotask,  │   │
│  │        MutationObserver, process.nextTick(Node)     │   │
│  │  调度：当前 macrotask 结束后立即清空全部微任务        │   │
│  │  特点：微任务中新增的微任务也会在本轮执行完毕          │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  requestAnimationFrame                                      │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  在渲染之前执行，不属于 microtask 也不属于 macrotask   │   │
│  │  执行时机：微任务清空后 → rAF 回调 → 样式计算 → 布局  │   │
│  │            → 绘制 → 下一个 macrotask                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

2.2 经典执行顺序示例

console.log('1: sync')

setTimeout(() => console.log('2: setTimeout'), 0)

Promise.resolve().then(() => console.log('3: Promise.then'))

queueMicrotask(() => console.log('4: queueMicrotask'))

requestAnimationFrame(() => console.log('5: rAF'))

console.log('6: sync end')

// 浏览器典型输出（单次 tick）：
// 1: sync
// 6: sync end
// 3: Promise.then
// 4: queueMicrotask
// 5: rAF (可能在下一帧)
// 2: setTimeout

2.3 setTimeout(fn, 0) 并非立即执行

延迟参数是最小等待时间，而非精确时间。setTimeout(fn, 0) 意味着"尽快将此回调放入 macrotask 队列"，但：

┌─────────────────────────────────────────────────────────────┐
│          为何 setTimeout(fn, 0) 不立即执行？                 │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  原因 1：嵌套层级限制                                        │
│  浏览器对嵌套超过 5 层的 setTimeout 强制最小 4ms 延迟        │
│                                                             │
│  原因 2：微任务优先                                          │
│  当前 macrotask 结束后必须先清空所有微任务                    │
│                                                             │
│  原因 3：渲染管线                                            │
│  浏览器可能在 macrotask 之间插入渲染帧（~16.6ms）            │
│                                                             │
│  原因 4：定时器合并                                          │
│  浏览器可能将相邻的多个定时器合并到同一时机执行               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

2.4 Node.js 特殊排序：nextTick vs microtask

// Node.js 环境中的优先级
Promise.resolve().then(() => console.log('Promise'))
process.nextTick(() => console.log('nextTick'))
queueMicrotask(() => console.log('queueMicrotask'))

// 输出顺序：
// nextTick        ← process.nextTick 优先级最高
// Promise         ← 然后是 Promise.then
// queueMicrotask  ← queueMicrotask 等同于 Promise.then

// 关键规则：Node.js 在每个事件循环阶段切换前，
// 先清空 nextTick 队列，再清空 microtask 队列

---

第3部分：Promise 深度解析

3.1 Promise 状态机

Promise 是不可逆的三态状态机：Pending → Fulfilled 或 Pending → Rejected。一旦 settled，状态永久锁定。

┌─────────────────────────────────────────────────────────────┐
│                    Promise 状态机                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│                   ┌──────────┐                              │
│                   │  Pending  │                              │
│                   │  (待定)    │                              │
│                   └─────┬─────┘                              │
│                         │                                    │
│            resolve(x)   │   reject(e) / throw                │
│                         │                                    │
│          ┌──────────────┼──────────────┐                    │
│          ▼              │              ▼                    │
│  ┌──────────────┐       │      ┌──────────────┐            │
│  │  Fulfilled   │       │      │   Rejected    │            │
│  │  (已兑现)     │       │      │   (已拒绝)     │            │
│  │  value: x    │       │      │  reason: e    │            │
│  └──────────────┘       │      └──────────────┘            │
│          │              │              │                    │
│          ▼              │              ▼                    │
│    .then(onFulfilled)   │     .then(null, onRejected)       │
│                         │     .catch(onRejected)            │
│                         │                                    │
│  规则：                                                      │
│  • Settled 后状态不可变                                      │
│  • .then() 永远返回一个新的 Promise                          │
│  • 回调返回值决定新 Promise 的状态                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

3.2 Promise 构造函数

// Promise 构造函数的执行是同步的
console.log('A')
new Promise((resolve, reject) => {
  console.log('B: executor 同步执行')
  resolve('done')
})
console.log('C')

// 输出：A → B → C
// executor 立即同步执行，.then 回调才是异步的

// 典型模式：封装回调风格的 API
function readFileAsync(path: string): Promise<string> {
  return new Promise((resolve, reject) => {
    fs.readFile(path, 'utf-8', (err, data) => {
      if (err) reject(err)
      else resolve(data)
    })
  })
}

3.3 链式调用与展平（Flattening）

// .then() 返回值自动展平
Promise.resolve('a')
  .then((val) => {
    // 返回普通值 → 包装为 Promise.resolve(value)
    return val + 'b'
  })
  .then((val) => {
    // 返回 Promise → 自动等待其 settled
    return new Promise((resolve) =>
      setTimeout(() => resolve(val + 'c'), 100)
    )
  })
  .then((val) => {
    console.log(val) // 'abc'
  })

// 错误传播：未被捕获的 rejection 沿链向下传播
Promise.resolve()
  .then(() => { throw new Error('oops') })
  .then(() => console.log('不会执行'))
  .catch((err) => console.log('捕获:', err.message))
  .then(() => console.log('catch 之后继续执行'))

3.4 六大静态方法决策树

┌─────────────────────────────────────────────────────────────┐
│              Promise 静态方法选择决策树                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  你有多个 Promise，想要...                                   │
│  │                                                          │
│  ├─ 全部成功才成功，任一失败则失败                            │
│  │  → Promise.all([p1, p2, p3])                            │
│  │                                                          │
│  ├─ 全部 settled 后获取结果（不论成功/失败）                  │
│  │  → Promise.allSettled([p1, p2, p3])                     │
│  │                                                          │
│  ├─ 第一个 settled 的结果（不论成功/失败）                    │
│  │  → Promise.race([p1, p2, p3])                           │
│  │                                                          │
│  ├─ 第一个 fulfilled 的结果，全部失败才 reject               │
│  │  → Promise.any([p1, p2, p3])                            │
│  │                                                          │
│  ├─ 需要在外部 resolve/reject 一个 Promise                   │
│  │  → Promise.withResolvers()  (ES2024)                    │
│  │                                                          │
│  └─ 包装已知值为 Promise                                     │
│     → Promise.resolve(val) / Promise.reject(err)            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

3.5 静态方法对比表

方法	输入	输出时机	失败行为
Promise.all	Iterable	全部 fulfilled	任一 rejected 立即 reject
Promise.allSettled	Iterable	全部 settled	永不 reject（结果含 status 字段）
Promise.race	Iterable	第一个 settled	第一个是 reject 则 reject
Promise.any	Iterable	第一个 fulfilled	全部 rejected 才抛 AggregateError
Promise.withResolvers	无	立即返回	手动控制

// Promise.allSettled 结果格式
const results = await Promise.allSettled([
  fetch('/api/users'),
  fetch('/api/orders'),
  Promise.reject(new Error('bad'))
])
// results[0]: { status: 'fulfilled', value: Response }
// results[1]: { status: 'fulfilled', value: Response }
// results[2]: { status: 'rejected', reason: Error('bad') }

// Promise.any 用于取最快成功的响应
const data = await Promise.any([
  fetch('/api/primary',  { signal: AbortSignal.timeout(500) }),
  fetch('/api/fallback', { signal: AbortSignal.timeout(800) }),
])

// Promise.withResolvers —— 解耦 Promise 创建与控制
const { promise, resolve, reject } = Promise.withResolvers<string>()
// 现在可以把 resolve/reject 传给回调，promise 传给调用方

3.6 AbortSignal.timeout 与 Promise.withResolvers

// AbortSignal.timeout() 简化超时处理（ES2024）
async function fetchWithTimeout(url: string, ms: number) {
  const res = await fetch(url, { signal: AbortSignal.timeout(ms) })
  return res.json()
}

// Promise.withResolvers 实现事件转 Promise
function waitForEvent(target: EventTarget, event: string) {
  const { promise, resolve } = Promise.withResolvers<Event>()
  target.addEventListener(event, resolve, { once: true })
  return promise
}

---

第4部分：async/await 内部机制

4.1 async 函数本质

async function 是返回 Promise 的语法糖。函数体内的 return 值自动包装为 Promise.resolve()，抛出的异常自动包装为 Promise.reject()。

// 这两种写法等价
async function f1(): Promise<number> {
  return 42
}

function f2(): Promise<number> {
  return Promise.resolve(42)
}
// 两者都返回 Promise<number>，而非 number

// 抛出异常等价
async function g1(): Promise<never> {
  throw new Error('fail')
}

function g2(): Promise<never> {
  return Promise.reject(new Error('fail'))
}

4.2 await 到底做了什么

await 将当前 async 函数的剩余部分作为 .then() 回调来调度。它暂停的是当前 async 函数的执行，而非整个线程。

┌─────────────────────────────────────────────────────────────┐
│                await 的底层等价变换                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  async function example() {                                 │
│    console.log('A')                                         │
│    const result = await fetch('/api')                       │
│    console.log('B', result)                                 │
│    return result                                            │
│  }                                                          │
│                                                             │
│  // 等价于：                                                 │
│  function example() {                                       │
│    console.log('A')                                         │
│    return fetch('/api').then((result) => {                  │
│      console.log('B', result)                               │
│      return result                                          │
│    })                                                       │
│  }                                                          │
│                                                             │
│  关键差异：                                                  │
│  • try/catch 在 async 中可捕获 rejected Promise             │
│  • 调试器可以像同步代码一样逐行调试                           │
│  • 但是 await 后的代码是微任务调度的                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

4.3 错误处理

// async 函数中 try/catch 可捕获 await 的 rejection
async function safeFetch(url: string) {
  try {
    const res = await fetch(url)
    if (!res.ok) throw new Error(`HTTP ${res.status}`)
    return await res.json()
  } catch (err) {
    console.error('请求失败:', err)
    return null // 降级处理
  }
}

// 并行操作的错误：每个独立处理
async function loadDashboard() {
  const [user, stats] = await Promise.all([
    safeFetch('/api/user').catch(() => null),
    safeFetch('/api/stats').catch(() => null),
  ])
  return { user, stats }
}

// 常见反模式：不必要的 try/catch 包装
// ❌ 错误
async function bad() {
  try {
    return await fetch('/api')
  } catch (e) {
    throw e  // 空捕获，等效于没有 try/catch
  }
}

// ✅ 正确：直接返回 Promise 让调用方处理
function good(): Promise<Response> {
  return fetch('/api')
}

4.4 await 在循环中：串行 vs 并行

// ❌ 串行执行：总耗时 = 每次请求耗时之和
async function fetchSequential(urls: string[]) {
  const results = []
  for (const url of urls) {
    results.push(await fetch(url)) // 每次等待前一个完成
  }
  return results
}

// ✅ 并行执行：总耗时 ≈ 最慢的一次请求
async function fetchParallel(urls: string[]) {
  const promises = urls.map((url) => fetch(url))
  return Promise.all(promises)
}

// ✅ 可控并行：限制并发数（见第7部分）
async function fetchWithLimit(urls: string[], limit: number) {
  const results: Response[] = []
  for (let i = 0; i < urls.length; i += limit) {
    const batch = urls.slice(i, i + limit).map((u) => fetch(u))
    results.push(...(await Promise.all(batch)))
  }
  return results
}

4.5 顶层 await（Top-level await）

// ES2022: 模块顶层可直接使用 await（ESM 模块）
// config.ts
const response = await fetch('/config.json')
export const config = await response.json()

// 注意：顶层 await 会阻塞依赖该模块的模块加载
// 仅在确实需要动态初始化时使用

---

第5部分：AbortController 与异步取消

5.1 取消机制全景

┌─────────────────────────────────────────────────────────────┐
│                AbortController 取消链路                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  AbortController                                            │
│  ┌──────────────────────┐                                   │
│  │  .signal: AbortSignal │──→ 传递给 fetch / 自定义操作     │
│  │  .abort(reason?)     │──→ 触发 signal.aborted = true    │
│  └──────────────────────┘                                   │
│           │                                                 │
│           ▼                                                 │
│  AbortSignal                                               │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  .aborted: boolean    — 是否已取消                   │   │
│  │  .reason: any         — 取消原因                     │   │
│  │  .throwIfAborted()   — 若已取消则抛出 reason         │   │
│  │  .onabort: handler   — 取消时触发                    │   │
│  │  addEventListener('abort', fn)                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  静态工厂方法：                                              │
│  • AbortSignal.timeout(ms) — 定时自动取消                   │
│  • AbortSignal.any([s1, s2]) — 任一取消则取消               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

5.2 取消 fetch 请求

async function fetchWithTimeout(
  url: string,
  timeoutMs: number
): Promise<Response> {
  const controller = new AbortController()
  const timer = setTimeout(() => controller.abort(), timeoutMs)

  try {
    const response = await fetch(url, { signal: controller.signal })
    return response
  } catch (err) {
    if ((err as Error).name === 'AbortError') {
      throw new Error(`请求超时 (${timeoutMs}ms): ${url}`)
    }
    throw err
  } finally {
    clearTimeout(timer)
  }
}

// 级联取消：用户取消 → 取消所有下游请求
function createCascadeAbort(parentSignal: AbortSignal) {
  const controller = new AbortController()
  // 组合信号：父取消或本地取消任一触发
  const combined = AbortSignal.any([parentSignal, controller.signal])
  // 监听父信号
  parentSignal.addEventListener('abort', () =>
    controller.abort(parentSignal.reason)
  )
  return { signal: combined, controller }
}

5.3 将 AbortSignal 集成到自定义异步操作

// 模式1: 轮询检查
async function delay(ms: number, signal?: AbortSignal): Promise<void> {
  return new Promise((resolve, reject) => {
    if (signal?.aborted) return reject(signal.reason)
    const timer = setTimeout(resolve, ms)
    signal?.addEventListener('abort', () => {
      clearTimeout(timer)
      reject(signal.reason)
    }, { once: true })
  })
}

// 模式2: 在循环中主动检查
async function processBatch(
  items: string[],
  signal?: AbortSignal
): Promise<void> {
  for (const item of items) {
    signal?.throwIfAborted()  // ES2024: 已取消则抛出
    await processItem(item)
  }
}

// 模式3: 将 AbortSignal 传递给底层 API
async function fetchLargeFile(url: string, signal?: AbortSignal) {
  const response = await fetch(url, { signal })
  const reader = response.body!.getReader()
  try {
    while (true) {
      const { done, value } = await reader.read()
      if (done) break
      signal?.throwIfAborted()
      // 处理 value (Uint8Array)
    }
  } finally {
    reader.releaseLock()
  }
}

---

第6部分：异步迭代器与生成器

6.1 异步迭代器协议

┌─────────────────────────────────────────────────────────────┐
│              同步迭代 vs 异步迭代                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  同步迭代器                 异步迭代器                        │
│  ──────────                 ──────────                       │
│  Symbol.iterator            Symbol.asyncIterator             │
│  → Iterator {              → AsyncIterator {                │
│      next(): {              next(): Promise<{               │
│        value, done           value, done                    │
│      }                        }>                            │
│    }                        }                               │
│                                                             │
│  for (const x of iter)     for await (const x of asyncIter) │
│                                                             │
└─────────────────────────────────────────────────────────────┘

6.2 异步生成器函数

// 异步生成器：async function* —— 可以 await 的生成器
async function* paginatedFetch<T>(
  baseUrl: string,
  pageSize: number
): AsyncGenerator<T[]> {
  let page = 0
  let hasMore = true
  while (hasMore) {
    const res = await fetch(`${baseUrl}?page=${page}&size=${pageSize}`)
    const data: { items: T[]; hasMore: boolean } = await res.json()
    yield data.items
    hasMore = data.hasMore
    page++
  }
}

// 消费
for await (const batch of paginatedFetch<User>('/api/users', 50)) {
  // 逐批处理，不会一次性加载全部数据
  await processBatch(batch)
}

// 手动迭代 + 取消
async function fetchAllUsers(signal?: AbortSignal) {
  const iter = paginatedFetch<User>('/api/users', 50)
  const all: User[] = []
  while (true) {
    signal?.throwIfAborted()
    const { value, done } = await iter.next()
    if (done) break
    all.push(...value)
  }
  return all
}

6.3 ReadableStream 与异步迭代

// ReadableStream 实现了 asyncIterator 协议（ES2023）
async function streamProcess(res: Response) {
  const decoder = new TextDecoder()
  for await (const chunk of res.body!) {
    // chunk 是 Uint8Array
    const text = decoder.decode(chunk, { stream: true })
    console.log('收到:', text)
  }
}

// 自定义 TransformStream + async iteration
function lineSplitter(): TransformStream<Uint8Array, string> {
  let buffer = ''
  const decoder = new TextDecoder()
  return new TransformStream({
    transform(chunk, controller) {
      buffer += decoder.decode(chunk, { stream: true })
      const lines = buffer.split('\n')
      buffer = lines.pop() || ''
      for (const line of lines) {
        controller.enqueue(line)
      }
    },
    flush(controller) {
      if (buffer) controller.enqueue(buffer)
    },
  })
}

// 使用
const response = await fetch('/api/stream')
const lines = response.body!.pipeThrough(lineSplitter())
for await (const line of lines) {
  console.log('行:', line)
}

---

第7部分：并发控制模式

7.1 Semaphore —— 限制并发数

┌─────────────────────────────────────────────────────────────┐
│              Semaphore 并发限制原理                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  并发上限 = 3                                                │
│                                                             │
│  任务队列：[T4, T5, T6, T7, T8, T9...]  ← 等待中            │
│                                                             │
│  执行中：  ┌────┐  ┌────┐  ┌────┐                          │
│            │ T1 │  │ T2 │  │ T3 │  ← 3 个槽位已满           │
│            └────┘  └────┘  └────┘                          │
│              │       │       │                               │
│              ▼       ▼       ▼                               │
│           完成 →  完成 →  完成 → 释放槽位 → T4 进入执行       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

class Semaphore {
  private permits: number
  private queue: (() => void)[] = []

  constructor(maxConcurrency: number) {
    this.permits = maxConcurrency
  }

  async acquire(): Promise<void> {
    if (this.permits > 0) {
      this.permits--
      return
    }
    return new Promise((resolve) => this.queue.push(resolve))
  }

  release(): void {
    if (this.queue.length > 0) {
      this.queue.shift()!()
    } else {
      this.permits++
    }
  }

  async run<T>(fn: () => Promise<T>): Promise<T> {
    await this.acquire()
    try {
      return await fn()
    } finally {
      this.release()
    }
  }
}

// 使用：最多 3 个并发请求
async function fetchAll(urls: string[], concurrency = 3) {
  const sem = new Semaphore(concurrency)
  return Promise.all(urls.map((url) => sem.run(() => fetch(url))))
}

7.2 批量任务调度器

// 通用模式：pMap —— 带并发限制的 Promise.all
async function pMap<T, R>(
  items: T[],
  fn: (item: T, index: number) => Promise<R>,
  concurrency: number = Infinity
): Promise<R[]> {
  const results: R[] = new Array(items.length)
  const sem = new Semaphore(concurrency)

  const tasks = items.map((item, i) =>
    sem.run(async () => {
      results[i] = await fn(item, i)
    })
  )

  await Promise.all(tasks)
  return results
}

// 使用示例：批量处理 100 个 URL，每次最多 5 个
const data = await pMap(
  urls,
  async (url, i) => {
    console.log(`处理第 ${i} 个`)
    return fetch(url).then((r) => r.json())
  },
  5
)

7.3 速率限制（Rate Limiting）

// Token Bucket 速率限制器
class RateLimiter {
  private tokens: number
  private lastRefill: number
  private readonly maxTokens: number
  private readonly refillRate: number // tokens per ms

  constructor(requestsPerSecond: number, burst?: number) {
    this.maxTokens = burst ?? requestsPerSecond
    this.tokens = this.maxTokens
    this.lastRefill = Date.now()
    this.refillRate = requestsPerSecond / 1000
  }

  async wait(): Promise<void> {
    this.refill()
    if (this.tokens >= 1) {
      this.tokens--
      return
    }
    const waitMs = Math.ceil((1 - this.tokens) / this.refillRate)
    await new Promise((r) => setTimeout(r, waitMs))
    this.tokens = 0
  }

  private refill(): void {
    const now = Date.now()
    const elapsed = now - this.lastRefill
    this.tokens = Math.min(
      this.maxTokens,
      this.tokens + elapsed * this.refillRate
    )
    this.lastRefill = now
  }
}

7.4 超时 + 重试 + 退避

async function fetchWithRetry(
  url: string,
  options: {
    retries?: number
    baseDelayMs?: number
    maxDelayMs?: number
    signal?: AbortSignal
  } = {}
): Promise<Response> {
  const { retries = 3, baseDelayMs = 1000, maxDelayMs = 30000, signal } = options

  for (let attempt = 0; attempt <= retries; attempt++) {
    try {
      const res = await fetch(url, { signal })
      if (res.ok) return res
      // 5xx 才重试，4xx 不重试
      if (res.status < 500 || attempt === retries) return res
    } catch (err) {
      if (attempt === retries || signal?.aborted) throw err
    }
    // 指数退避 + 随机抖动
    const delay = Math.min(
      baseDelayMs * Math.pow(2, attempt) + Math.random() * 1000,
      maxDelayMs
    )
    await new Promise((r) => setTimeout(r, delay))
  }
  throw new Error('unreachable')
}

---

第8部分：常见异步陷阱

8.1 Floating Promise（遗忘 await）

// ❌ 错误：Promise 没有被 await，错误被静默吞掉
async function handleClick() {
  saveToDatabase(data) // 返回 Promise，但未 await！
  redirectTo('/dashboard') // 立即跳转，save 可能未完成
}

// ✅ 正确：要么 await，要么显式处理
async function handleClick() {
  await saveToDatabase(data)
  redirectTo('/dashboard')
}

// ✅ 备选：fire-and-forget 时也要处理错误
function handleClick() {
  saveToDatabase(data).catch((err) => {
    console.error('保存失败', err)
    showErrorToast('保存失败，请重试')
  })
  redirectTo('/dashboard')
}

8.2 async forEach 陷阱

// ❌ 错误：forEach 不回等待 async 回调
async function processAll(items: string[]) {
  items.forEach(async (item) => {
    await processItem(item) // 这些 Promise 全被丢弃！
  })
  console.log('done') // 在所有 processItem 完成之前就打印了
}

// ✅ 方案1：for...of 循环
async function processAll(items: string[]) {
  for (const item of items) {
    await processItem(item)
  }
  console.log('done') // 所有处理完成后才打印
}

// ✅ 方案2：并行 + Promise.all
async function processAll(items: string[]) {
  await Promise.all(items.map((item) => processItem(item)))
  console.log('done')
}

8.3 Promise 链中忘记 return

// ❌ 错误：then 中没有 return，导致链断裂
fetch('/api/user')
  .then((res) => {
    res.json() // 应该 return res.json()！
  })
  .then((user) => {
    console.log(user) // undefined！
  })

// ✅ 正确：必须 return
fetch('/api/user')
  .then((res) => res.json())
  .then((user) => console.log(user))

// ✅ 或使用 async/await 彻底避免此问题
async function getUser() {
  const res = await fetch('/api/user')
  return res.json()
}

8.4 异步竞态条件

// ❌ 错误：竞态条件 —— 旧请求可能覆盖新请求
let lastQuery = ''
async function search(query: string) {
  lastQuery = query
  const results = await fetch(`/api/search?q=${query}`)
  if (query !== lastQuery) return // 过期结果，丢弃！
  renderResults(await results.json())
}

// ✅ 使用 AbortController 取消旧请求
let abortController: AbortController | null = null
async function search(query: string) {
  abortController?.abort()
  abortController = new AbortController()
  try {
    const results = await fetch(`/api/search?q=${query}`, {
      signal: abortController.signal,
    })
    renderResults(await results.json())
  } catch (err) {
    if ((err as Error).name !== 'AbortError') throw err
  }
}

8.5 Promisify 陷阱

// ❌ 错误：忘记处理 error 事件
function badPromisify(stream: EventEmitter): Promise<Buffer> {
  return new Promise((resolve) => {
    const chunks: Buffer[] = []
    stream.on('data', (chunk) => chunks.push(chunk))
    stream.on('end', () => resolve(Buffer.concat(chunks)))
    // 缺少 stream.on('error', reject)！
  })
}

// ✅ 正确：完整的事件处理
function goodPromisify(stream: EventEmitter): Promise<Buffer> {
  return new Promise((resolve, reject) => {
    const chunks: Buffer[] = []
    stream.on('data', (chunk) => chunks.push(chunk))
    stream.on('end', () => resolve(Buffer.concat(chunks)))
    stream.on('error', reject)
  })
}

8.6 混用同步与异步的控制流

// ❌ 陷阱：条件分支中混用 async
function getConfig(): Config | Promise<Config> {
  if (cachedConfig) {
    return cachedConfig // 同步返回
  }
  return fetch('/config').then((r) => r.json()) // 异步返回
}
// 调用者不知道该 await 还是直接用！

// ✅ 始终返回 Promise
async function getConfig(): Promise<Config> {
  if (cachedConfig) return cachedConfig
  const res = await fetch('/config')
  cachedConfig = await res.json()
  return cachedConfig
}

8.7 Unhandled Rejection 与全局监听

// 未处理的 Promise rejection 会导致 Node.js 进程崩溃
// 但不会像同步异常那样立即崩溃

// 全局捕获（防御性编程）
process.on('unhandledRejection', (reason, promise) => {
  console.error('未处理的 Promise rejection:', reason)
  // 记录日志、发送告警，但不要阻止进程退出
})

// 浏览器端
window.addEventListener('unhandledrejection', (event) => {
  console.error('未处理的 rejection:', event.reason)
  event.preventDefault() // 阻止默认的控制台错误
})

---

第9部分：Streams API 概述

9.1 三种核心流类型

┌─────────────────────────────────────────────────────────────┐
│                    Streams API 核心概念                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ReadableStream<T>          WritableStream<T>               │
│  ┌──────────────────┐      ┌──────────────────┐            │
│  │ reader.read()     │      │ writer.write()    │            │
│  │      ↓            │      │      ↓            │            │
│  │ { value, done }   │      │ writer.close()    │            │
│  │ controller.enqueue│      │ controller.error  │            │
│  └──────────────────┘      └──────────────────┘            │
│           │                         ▲                       │
│           │    ┌──────────────┐     │                       │
│           └───→│TransformStream│────┘                       │
│                │ readable      │                             │
│                │ writable      │                             │
│                │ .transform()  │                             │
│                └──────────────┘                             │
│                                                             │
│  关键方法：                                                  │
│  • readable.pipeThrough(transform) → ReadableStream         │
│  • readable.pipeTo(writable) → Promise<void>                │
│  • readable.tee() → [ReadableStream, ReadableStream]        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

9.2 fetch response.body 流式读取

// 以 ReadableStream 方式读取大文件
async function downloadProgressively(url: string) {
  const response = await fetch(url)
  const contentLength = Number(response.headers.get('Content-Length')) || 0
  const reader = response.body!.getReader()
  let received = 0
  const chunks: Uint8Array[] = []

  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    chunks.push(value)
    received += value.length
    console.log(`进度: ${((received / contentLength) * 100).toFixed(1)}%`)
  }

  return new Blob(chunks)
}

// tee() 分支：一份数据两个消费者（如边下载边 hash）
async function downloadAndHash(url: string) {
  const response = await fetch(url)
  const [stream1, stream2] = response.body!.tee()

  // 分支1：保存到文件
  const savePromise = saveToFile(stream1)

  // 分支2：计算 SHA-256
  const hashPromise = computeSHA256(stream2)

  const [path, hash] = await Promise.all([savePromise, hashPromise])
  return { path, hash }
}

9.3 自定义 ReadableStream

// 生成一个"计时器流"，每秒产生一个时间戳
function createTimerStream(count: number): ReadableStream<string> {
  let emitted = 0
  return new ReadableStream({
    async pull(controller) {
      if (emitted >= count) {
        controller.close()
        return
      }
      await new Promise((r) => setTimeout(r, 1000))
      controller.enqueue(new Date().toISOString())
      emitted++
    },
  })
}

for await (const timestamp of createTimerStream(5)) {
  console.log('滴答:', timestamp)
}

---

核心总结

1. 事件循环是 JavaScript 异步的心脏

事件循环遵循"一个 macrotask → 清空所有 microtask → 可能渲染 → 下一个 macrotask"的核心循环。理解这个模型是写出正确异步代码的前提。Node.js 的 libuv 六阶段模型是其变体，process.nextTick 在每个阶段切换前优先执行。

2. Promise 是构建异步代码的基石

Promise 是不可逆的三态状态机。.then() 永远返回新 Promise，错误沿链传播。选择合适的静态方法（all、allSettled、race、any）是高效并发编程的关键。async/await 让异步代码看起来像同步代码，但本质仍是 Promise + 微任务调度。

3. 并发控制与取消是生产系统的必备能力

Promise.all 对大量任务不设限可能耗尽连接和内存。Semaphore 模式提供可控并发，AbortController 提供级联取消。AbortSignal.timeout() 和 AbortSignal.any() 使超时和组合取消变得简单。异步迭代器和 Streams API 让流式数据处理有了标准化方案。

---

章节测试

Q1: 以下代码的输出顺序是？

console.log('1')
setTimeout(() => console.log('2'), 0)
Promise.resolve().then(() => console.log('3'))
queueMicrotask(() => {
  console.log('4')
  Promise.resolve().then(() => console.log('5'))
})
console.log('6')

A) 1, 6, 3, 4, 2, 5 B) 1, 6, 3, 4, 5, 2 C) 1, 2, 3, 4, 5, 6 D) 1, 6, 4, 3, 5, 2

Q2: Promise.all 与 Promise.allSettled 的核心区别是什么？各自适合什么场景？

Q3: 以下代码有什么问题？如何修复？

async function process(items: number[]) {
  items.forEach(async (item) => {
    await save(item)
  })
  console.log('all done')
}

Q4: async function 的返回值类型是什么？为什么 async function f() { return 42 } 的返回类型是 Promise<number> 而非 number？

Q5: Node.js 中 process.nextTick、Promise.then、setImmediate 三者的执行顺序是怎样的？

Q6: 如何用 AbortController 实现一个 5 秒超时的 fetch，且超时后不要留下未清理的 timer？

Q7: 补全以下代码，实现最多 3 个并发请求的批量下载功能：

async function downloadAll(urls: string[]): Promise<Response[]> {
  // 你的实现
}

---

参考答案

A1: B。同步代码先执行 (1, 6)；然后是微任务清空阶段：先 3 (Promise.then)，再 4 (queueMicrotask)，queueMicrotask 内部新增的 5 也在同一轮清空；最后是宏任务 2 (setTimeout)。

A2: Promise.all 在任一输入 reject 时立即 reject（fail-fast），适合"全部成功才有意义"的场景。Promise.allSettled 等待所有输入 settled，结果数组包含 {status, value/reason}，适合"尽可能多地获取结果，容忍部分失败"的场景。

A3: forEach 不会等待 async 回调，'all done' 在所有 save() 完成前就打印了。修复：使用 for...of 循环逐次 await，或 await Promise.all(items.map(save)) 并行执行。

A4: async function 始终返回 Promise。return 42 等价于 return Promise.resolve(42)。调用方需要 .then() 或 await 才能拿到 42。这是 async 函数的核心约定。

A5: 执行顺序为：process.nextTick (最高优先) → Promise.then / queueMicrotask → setImmediate (下一轮事件循环的 check 阶段)。nextTick 在每个阶段切换前执行，microtask 在 nextTick 之后执行，setImmediate 在 poll 阶段之后的 check 阶段执行。

A6:

async function fetchWith5sTimeout(url: string) {
  const controller = new AbortController()
  const timer = setTimeout(() => controller.abort(), 5000)
  try {
    return await fetch(url, { signal: controller.signal })
  } finally {
    clearTimeout(timer) // 成功或失败都清理 timer
  }
}

A7: 参见第7部分的 pMap 或 Semaphore 实现，核心是控制并发数为 3。

---

相关笔记

• [[01-javascript-runtime-and-language]] — JavaScript 运行时环境、V8 引擎、内存模型
• [[02-typescript-type-system]] — TypeScript 中 Promise 泛型标注与类型推导
• [[00-overview]] — JavaScript 与 TypeScript 章节总览

---

下一步学习

• [ ] 在浏览器 DevTools 的 Performance 面板中录制并观察事件循环的行为
• [ ] 手写一个 Semaphore 类并在实际项目中使用它控制并发
• [ ] 将项目中所有的"遗忘 await"和"async forEach"问题修复
• [ ] 用 AbortController 改造现有请求管理，实现级联取消
• [ ] 尝试用 ReadableStream 和 TransformStream 处理流式数据

学习状态：🟡 开始学习