Node.js 异步模式与错误处理 / Async Patterns and Error Handling in Node.js

📅 创建时间：2026-07-28 🏷️ 标签：#Nodejs #AsyncPatterns #ErrorHandling #GracefulShutdown 📚 前置知识：[[01-nodejs-runtime-internals]]

---

📋 本章目标

• 理解 Node.js 异步模型的演进（callback → Promise → async/await）
• 掌握 async/await 的内部机制与常见陷阱
• 能够设计统一的错误处理策略（操作错误 vs 程序错误）
• 掌握并发控制模式（信号量、节流、批处理）
• 实现优雅关闭（graceful shutdown）与健康检查
• 建立结构化日志体系

---

第1部分：异步模型的演进

1.1 为什么 Node.js 必须异步

┌─────────────────────────────────────────────────────────────┐
│                    同步 vs 异步——一个请求的代价                │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  同步模型（每个连接一个线程）：                               │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ Request 1: [====== 等待 DB 响应 ======]              │   │
│  │ Request 2:                  [====== 等待 DB ======]  │   │
│  │ Request 3:                                   [==]   │   │
│  │                                                      │   │
│  │ 线程大部分时间在等待 → 资源浪费                       │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  Node.js 异步模型（一个线程 + 事件循环）：                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ Request 1: [req]==> [register callback] 等待...     │   │
│  │ Request 2:        [req]==> [register callback] 等待  │   │
│  │ Request 3:                 [req]==> [callback runs]  │   │
│  │                                                      │   │
│  │ 单线程处理所有请求 → 等待期间可处理其他工作           │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.2 三代异步模式的进化

┌─────────────────────────────────────────────────────────────┐
│                    异步模式三代演进                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  第一代：Callback（回调地狱）                                │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ fs.readFile('a.txt', (err, dataA) => {              │   │
│  │   fs.readFile('b.txt', (err, dataB) => {            │   │
│  │     fs.readFile('c.txt', (err, dataC) => {          │   │
│  │       // 回调地狱：代码向右增长                      │   │
│  │     })                                              │   │
│  │   })                                                │   │
│  │ })                                                  │   │
│  │ 核心问题：错误处理分散、控制流难以追踪               │   │
│  └─────────────────────────────────────────────────────┘   │
│                            ↓                                │
│  第二代：Promise（链式调用）                                 │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ fs.promises.readFile('a.txt')                       │   │
│  │   .then(dataA => fs.promises.readFile('b.txt'))     │   │
│  │   .then(dataB => fs.promises.readFile('c.txt'))     │   │
│  │   .catch(err => console.error(err))                 │   │
│  │ 改进：统一的错误处理、可组合                           │   │
│  └─────────────────────────────────────────────────────┘   │
│                            ↓                                │
│  第三代：async/await（同步风格）                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ try {                                               │   │
│  │   const dataA = await fs.promises.readFile('a.txt') │   │
│  │   const dataB = await fs.promises.readFile('b.txt') │   │
│  │   const dataC = await fs.promises.readFile('c.txt') │   │
│  │ } catch (err) {                                     │   │
│  │   console.error(err)                                │   │
│  │ }                                                   │   │
│  │ 改进：自然的 try/catch、可调试的调用栈               │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.3 async/await 的内部机制

// async 函数 = 返回 Promise 的普通函数
async function getUser(id: number): Promise<User> {
  // await = .then() 的语法糖
  // await 之后的代码 = .then() 中的回调
  const user = await db.user.findUnique({ where: { id } })
  return user
}

// 上面的代码等价于：
function getUser(id: number): Promise<User> {
  return db.user.findUnique({ where: { id } })
    .then(user => user)
}

// 关键理解：await 不会阻塞事件循环
// 它让当前 async 函数"暂停"并将控制权交还给事件循环
// 当 Promise resolve 后，函数从暂停处继续执行

┌─────────────────────────────────────────────────────────────┐
│                    await 的执行模型                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  async function example() {                                 │
│    console.log('A')                                         │
│    const result = await fetchData()  ← 在此"暂停"          │
│    console.log('B')                ← Promise resolve 后继续  │
│  }                                                          │
│                                                             │
│  时间线：                                                    │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 调用 example()                                       │   │
│  │   → 执行 console.log('A')                           │   │
│  │   → 遇到 await：将 fetchData() 加入微任务队列       │   │
│  │   → 函数返回一个未完成的 Promise                    │   │
│  │   → 事件循环继续处理其他任务                        │   │
│  │   → fetchData() resolve 后：                        │   │
│  │   → 微任务队列调度：执行 console.log('B')           │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ⚠️ await 后面的代码被推入微任务队列，不阻塞主线程          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

第2部分：错误处理策略

2.1 操作错误 vs 程序错误

┌─────────────────────────────────────────────────────────────┐
│                    两类错误的处理哲学                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  操作错误（Operational Errors）：                            │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ • 数据库连接超时                                     │   │
│  │ • 用户输入格式错误                                   │   │
│  │ • 第三方 API 返回 500                                │   │
│  │ • 文件不存在                                         │   │
│  │                                                      │   │
│  │ 处理策略：捕获 → 记录 → 返回有意义的错误响应         │   │
│  │ 程序应继续运行                                       │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  程序错误（Programmer Errors）：                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ • 读取 undefined 的属性                              │   │
│  │ • 传递给函数的参数类型错误                           │   │
│  │ • Promise 没有 .catch 处理                           │   │
│  │                                                      │   │
│  │ 处理策略：让它 crash → 重启                          │   │
│  │ 这些是 bug，不应试图在运行时修复                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

2.2 统一错误处理中间件

// 自定义错误类层次
class AppError extends Error {
  constructor(
    message: string,
    public readonly statusCode: number,
    public readonly code: string,
    public readonly isOperational: boolean = true
  ) {
    super(message)
    this.name = 'AppError'
    Error.captureStackTrace(this, this.constructor)
  }
}

class NotFoundError extends AppError {
  constructor(resource: string) {
    super(`${resource} not found`, 404, 'NOT_FOUND')
  }
}

class ValidationError extends AppError {
  constructor(public readonly details: unknown) {
    super('Validation failed', 400, 'VALIDATION_ERROR')
  }
}

// Express 错误处理中间件（4 参数签名）
function errorHandler(
  err: Error,
  req: Request,
  res: Response,
  next: NextFunction
) {
  // 操作错误：返回结构化响应
  if (err instanceof AppError && err.isOperational) {
    return res.status(err.statusCode).json({
      error: {
        code: err.code,
        message: err.message,
        ...(err instanceof ValidationError && { details: err.details }),
      }
    })
  }

  // 程序错误：记录详情，返回通用 500
  logger.error('Unexpected error', { err, requestId: req.id })
  return res.status(500).json({
    error: {
      code: 'INTERNAL_ERROR',
      message: 'An unexpected error occurred'
    }
  })
}

2.3 异步错误传播链

┌─────────────────────────────────────────────────────────────┐
│                    Express 错误传播路径                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 同步错误：                                          │   │
│  │ throw new Error('...')     → Express 自动捕获       │   │
│  │                             → 跳转到错误处理中间件  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 异步错误（Express 4）：                             │   │
│  │ async (req, res, next) => {                        │   │
│  │   throw new Error('...') // ❌ Express 4 不会捕获   │   │
│  │ }                                                   │   │
│  │ → 必须显式调用 next(err)                           │   │
│  │ → 或使用 express-async-errors 补丁                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 异步错误（Express 5）：                             │   │
│  │ async (req, res, next) => {                        │   │
│  │   throw new Error('...') // ✅ Express 5 自动捕获   │   │
│  │ }                                                   │   │
│  │ → 自动传递 rejected Promise 到错误处理中间件        │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

第3部分：并发控制模式

3.1 有限并发执行

// 信号量模式：限制同时进行的异步操作数量
async function asyncPool<T, R>(
  concurrency: number,
  items: T[],
  fn: (item: T) => Promise<R>
): Promise<R[]> {
  const results: R[] = []
  const executing = new Set<Promise<void>>()

  for (const [index, item] of items.entries()) {
    const promise = fn(item).then(result => {
      results[index] = result
    })

    executing.add(promise)
    const cleanup = () => executing.delete(promise)
    promise.then(cleanup, cleanup)

    if (executing.size >= concurrency) {
      await Promise.race(executing)
    }
  }

  await Promise.all(executing)
  return results
}

// 使用示例：最多同时 3 个请求
const urls = ['url1', 'url2', 'url3', 'url4', 'url5']
const responses = await asyncPool(3, urls, fetchUrl)

┌─────────────────────────────────────────────────────────────┐
│                    并发控制示意图                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  输入队列: [task1, task2, task3, task4, task5]              │
│  并发限制: 3                                                │
│                                                             │
│  时间线:                                                     │
│  t0: [task1] [task2] [task3] → 同时启动                    │
│  t1: task1 完成 → [task4] 启动                              │
│  t2: task3 完成 → [task5] 启动                              │
│  t3: task2 完成                                             │
│  t4: task4 完成                                             │
│  t5: task5 完成                                             │
│                                                             │
│  任何时候最多只有 3 个任务在执行                             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

3.2 常见的 async/await 陷阱

// ❌ 陷阱1：forEach 中的 async 不会等待
async function processItems(items: Item[]) {
  items.forEach(async (item) => {
    await saveItem(item)  // forEach 不等待这个 Promise！
  })
  console.log('Done')  // 可能在 saveItem 完成前就打印了
}

// ✅ 正确写法：使用 for...of
async function processItems(items: Item[]) {
  for (const item of items) {
    await saveItem(item)
  }
  console.log('Done')  // 所有 item 保存完成后才打印
}

// ✅ 如果需要并行：使用 Promise.all
async function processItems(items: Item[]) {
  await Promise.all(items.map(item => saveItem(item)))
  console.log('Done')
}

// ❌ 陷阱2：忘记 await
async function getData() {
  const user = db.user.findUnique({ where: { id: 1 } })
  // 忘记 await → user 是 Promise<User>，不是 User
  return user.name  // undefined！因为 Promise 没有 name 属性
}

// ❌ 陷阱3：串行执行可并行的操作
const user = await getUser(id)        // 等待
const posts = await getPosts(id)      // 等待（可以并行！）
const settings = await getSettings(id) // 等待（可以并行！）

// ✅ 并行执行
const [user, posts, settings] = await Promise.all([
  getUser(id),
  getPosts(id),
  getSettings(id),
])

// ❌ 陷阱4：忘记处理 rejected Promise
async function handler() {
  updateCache(userId) // 返回 Promise，但没有 await 或 .catch
  // 如果 updateCache 失败，错误会被静默吞掉
  // → 触发 unhandledRejection 事件
}

---

第4部分：优雅关闭与健康检查

4.1 优雅关闭流程

┌─────────────────────────────────────────────────────────────┐
│                    优雅关闭（Graceful Shutdown）              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  收到 SIGTERM / SIGINT                                       │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 1. 停止接收新请求                                    │   │
│  │    server.close() → 不再接受新连接                   │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 2. 等待现有请求完成（设置超时上限）                  │   │
│  │    http-terminator / stoppable 库                    │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 3. 关闭数据库连接池                                  │   │
│  │    await prisma.$disconnect()                       │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 4. 关闭消息队列消费者                                │   │
│  │    await kafka.consumer.disconnect()                 │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 5. 刷新日志缓冲区                                    │   │
│  │    await logger.flush()                             │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 6. process.exit(0) 或 process.exit(1)               │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ⚠️ 必须设置强制超时：如果 30s 内未完成则 process.exit(1)  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

// 优雅关闭实现
import { createTerminus } from '@godaddy/terminus'

function createGracefulShutdown(server: Server) {
  createTerminus(server, {
    // 优雅关闭前的准备时间
    timeout: 30000,

    // 就绪检查（Kubernetes readiness probe）
    healthChecks: {
      '/health': async () => {
        // 检查关键依赖是否可用
        await db.$queryRaw`SELECT 1`
        await redis.ping()
      },
      '/live': async () => {
        // 仅检查进程是否存活
      }
    },

    // 收到关闭信号时执行
    beforeShutdown: async () => {
      logger.info('Server is shutting down...')
    },

    // 等待现有请求完成后的清理
    onShutdown: async () => {
      await db.$disconnect()
      await redis.quit()
      logger.info('Cleanup complete, exiting')
    },

    // 信号监听
    signals: ['SIGTERM', 'SIGINT'],
  })
}

4.2 健康检查端点设计

┌─────────────────────────────────────────────────────────────┐
│                    健康检查端点设计                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  /health（就绪探针 Readiness Probe）:                        │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 检查所有关键依赖：                                   │   │
│  │ • 数据库连接（SELECT 1）                             │   │
│  │ • Redis 连接（PING）                                 │   │
│  │ • 消息队列连接                                       │   │
│  │ • 磁盘空间（如需要）                                 │   │
│  │                                                      │   │
│  │ 返回 200：服务就绪，可接收流量                       │   │
│  │ 返回 503：依赖不可用，从负载均衡中移除               │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  /live（存活探针 Liveness Probe）:                           │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 仅检查进程是否存活：                                 │   │
│  │ • 返回 200 即表示存活                               │   │
│  │ • 不应检查外部依赖（避免级联重启）                   │   │
│  │                                                      │   │
│  │ 返回 200：进程存活                                   │   │
│  │ 超时/无响应：K8s 将重启 Pod                         │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

第5部分：结构化日志

5.1 为什么需要结构化日志

┌─────────────────────────────────────────────────────────────┐
│                    文本日志 vs 结构化日志                     │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  文本日志（难以查询）：                                      │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ User 42 purchased item "laptop" for $999 at 14:30   │   │
│  │                                                      │   │
│  │ 问题："查出所有购买金额 > $500 的用户" → 无法查询     │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  结构化日志（可查询、可聚合）：                              │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ {                                                   │   │
│  │   "level": "info",                                 │   │
│  │   "message": "Purchase completed",                  │   │
│  │   "userId": 42,                                    │   │
│  │   "item": "laptop",                                │   │
│  │   "amount": 999,                                   │   │
│  │   "timestamp": "2026-07-28T14:30:00Z"              │   │
│  │ }                                                   │   │
│  │                                                      │   │
│  │ 查询："amount > 500" → 瞬间完成，可生成图表          │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

5.2 pino 日志实践

import pino from 'pino'

const logger = pino({
  level: process.env.LOG_LEVEL ?? 'info',
  // 生产环境用 JSON 格式，开发环境用可读格式
  transport: process.env.NODE_ENV === 'development'
    ? { target: 'pino-pretty', options: { colorize: true } }
    : undefined,
  // 自动包含的字段
  base: {
    service: 'user-api',
    version: process.env.APP_VERSION,
  },
  // 自动生成 requestId
  mixin() {
    return {}
  },
})

// 每个请求注入 requestId
app.use((req, res, next) => {
  req.id = req.headers['x-request-id'] as string ?? crypto.randomUUID()
  req.log = logger.child({ requestId: req.id })
  next()
})

// 使用
app.get('/users/:id', async (req, res) => {
  req.log.info({ userId: req.params.id }, 'Fetching user')
  const user = await db.user.findUnique({ where: { id: req.params.id } })
  req.log.info({ userId: user?.id }, 'User fetched')
  res.json(user)
})

---

核心总结

总结1：异步模式的本质

async/await 没有改变 Node.js 的单线程事件循环模型——它只是让异步代码看起来像同步代码。await 之后的代码会作为微任务被调度，不阻塞事件循环。理解这点才能理解为什么 await 在循环中可能导致性能问题（应该用 Promise.all 并行化）。

总结2：错误处理的分层

操作错误（网络超时、验证失败）→ 捕获、记录、优雅返回。程序错误（类型错误、逻辑错误）→ 让它 crash，依赖进程管理器重启。不要在 catch 块里试图修复 bug——bug 应该在开发和测试阶段发现。

总结3：生产环境的必备实践

1. 每个请求注入 requestId → 贯穿所有日志
2. 实现健康检查端点（/health + /live）
3. 实现优雅关闭（30s 超时）
4. 使用结构化日志（JSON 格式，可被 ELK/Datadog 索引）

---

章节测试

测试1：await 后面的代码在什么时机执行？

A. 同步立即执行 B. 作为宏任务执行 C. 作为微任务执行 D. 在下一个事件循环 tick 执行

测试2：以下代码的输出顺序是什么？

async function test() {
  console.log('A')
  await Promise.resolve()
  console.log('B')
}
test()
console.log('C')

A. A, B, C B. A, C, B C. C, A, B D. B, A, C

测试3：为什么 forEach 不能与 async/await 正确配合？

测试4：Liveness Probe 和 Readiness Probe 的区别是什么？

测试5：操作错误和程序错误应该如何处理？

---

参考答案

测试1答案

答案：C。await 后的代码作为微任务（microtask）被调度——等价于 Promise.resolve().then(() => { ... })。

测试2答案

答案：B。执行顺序：A（同步）→ await 将 B 推入微任务队列 → C（同步）→ 微任务队列清空 → B。

测试3答案

forEach 的回调是同步执行的，它不会等待 async 回调返回的 Promise。即使回调内部有 await，forEach 也会立即继续到下一次迭代和后续代码。应该使用 for...of 或 Promise.all(items.map(...))。

测试4答案

Liveness Probe（/live）：仅检查进程是否存活，不应检查外部依赖（避免级联重启）。失败 → K8s 重启 Pod。 Readiness Probe（/health）：检查所有关键依赖是否可用（数据库、Redis 等）。失败 → 从负载均衡中移除，但不重启 Pod。恢复后 → 重新加入负载均衡。

测试5答案

操作错误（网络超时、验证失败、文件不存在）：是程序正常运行中可预期的问题 → 捕获、记录、返回有意义的错误响应给用户。 程序错误（类型错误、读取 undefined 的属性）：是代码中的 bug → 应该让它 crash，依赖进程管理器（如 PM2/K8s）重启。不应在运行时尝试修复 bug。

---

相关笔记

• [[01-nodejs-runtime-internals]] — libuv 事件循环与 V8 深度
• [[03-express-deep-dive]] — Express 中间件与请求生命周期
• [[../07-middleware/00-overview]] — 中间件生态

---

下一步学习

• [ ] 阅读 Express 深度解析 — 中间件洋葱模型
• [ ] 在你的 Express 应用中实现 requestId 注入和结构化日志
• [ ] 为你的服务添加 /health 和 /live 端点
• [ ] 用 k6 或 autocannon 压测你的服务，观察并发行为

学习状态：🟡 开始学习