Node.js 异步编程与并发控制实战 2026 | Event Loop 完全指南

Node.js 的异步特性是其高性能的核心，但也是最容易出错的地方。理解 Event Loop 原理和掌握并发控制技巧，是编写高质量 Node.js 代码的关键。本文将从 Event Loop 深入到各种异步模式，再到实战级并发控制方案。

---

一、Event Loop 原理深度解析

1.1 Event Loop 阶段

┌─────────────────────────────────────────────────────────────┐
│                    Event Loop 执行顺序                      │
├─────────────────────────────────────────────────────────────┤
│  1. timers          → setTimeout/setInterval 回调          │
│  2. pending callbacks → I/O 回调（除 close、timer、setImmediate）│
│  3. idle, prepare   → 内部使用                             │
│  4. poll            → 轮询新的 I/O 事件                    │
│     ├─ 执行 poll 队列中的回调                              │
│     └─ 如果 timers 队列有任务，跳到 timers                  │
│  5. check           → setImmediate 回调                   │
│  6. close callbacks → close 事件回调                       │
│     └─ 回到 timers（循环）                                │
└─────────────────────────────────────────────────────────────┘

1.2 微任务与宏任务

// 微任务队列（Microtask Queue）
Promise.resolve().then(() => console.log('micro 1'))
queueMicrotask(() => console.log('micro 2'))

// 宏任务队列（Macrotask Queue）
setTimeout(() => console.log('macro 1'), 0)
setImmediate(() => console.log('macro 2'))

// 执行顺序：所有微任务 → 下一个宏任务 → 所有微任务 → ...

微任务优先级高于宏任务，在每个阶段结束时都会清空微任务队列。

1.3 setTimeout vs setImmediate

// 在主模块中执行，顺序不确定
setTimeout(() => console.log('timeout'), 0)
setImmediate(() => console.log('immediate'))

// 在 I/O 回调中执行，setImmediate 总是先执行
fs.readFile('/etc/hosts', () => {
  setTimeout(() => console.log('timeout'), 0)
  setImmediate(() => console.log('immediate'))  // 先执行
})

1.4 process.nextTick 特殊地位

process.nextTick 不属于任何阶段，它在每个阶段切换之前执行：

Promise.resolve().then(() => console.log('promise'))
process.nextTick(() => console.log('nextTick'))  // 最先执行

// 输出：nextTick → promise

---

二、异步编程模式演进

2.1 回调地狱（Callback Hell）

fs.readFile('config.json', (err, data) => {
  if (err) throw err
  const config = JSON.parse(data)
  
  db.connect(config.db, (err, conn) => {
    if (err) throw err
    
    conn.query('SELECT * FROM users', (err, users) => {
      if (err) throw err
      
      users.forEach(user => {
        sendEmail(user.email, (err) => {
          if (err) console.error(err)
        })
      })
    })
  })
})

2.2 Promise 模式

const readFile = promisify(fs.readFile)

readFile('config.json', 'utf8')
  .then(data => JSON.parse(data))
  .then(config => db.connect(config.db))
  .then(conn => conn.query('SELECT * FROM users'))
  .then(users => Promise.all(
    users.map(user => sendEmail(user.email))
  ))
  .catch(err => console.error(err))

2.3 async/await 模式

async function processUsers() {
  try {
    const data = await readFile('config.json', 'utf8')
    const config = JSON.parse(data)
    const conn = await db.connect(config.db)
    const users = await conn.query('SELECT * FROM users')
    
    await Promise.all(users.map(user => sendEmail(user.email)))
    
    return users.length
  } catch (err) {
    console.error('处理失败:', err)
    throw err
  }
}

processUsers().then(count => console.log(`处理了 ${count} 个用户`))

2.4 Generator + co 模式

const co = require('co')

co(function* () {
  const data = yield readFile('config.json', 'utf8')
  const config = JSON.parse(data)
  const conn = yield db.connect(config.db)
  const users = yield conn.query('SELECT * FROM users')
  
  yield users.map(user => sendEmail(user.email))
  return users.length
}).then(count => console.log(`处理了 ${count} 个用户`))

---

三、Promise 进阶技巧

3.1 Promise.all / allSettled / race / any

// Promise.all：全部成功，任一失败则失败
const results = await Promise.all([
  fetchData('users'),
  fetchData('posts'),
  fetchData('comments')
])

// Promise.allSettled：全部完成，返回每个的状态
const results = await Promise.allSettled([
  fetchData('users'),
  fetchData('posts'),
  fetchData('comments')
])
results.forEach(result => {
  if (result.status === 'fulfilled') {
    console.log('成功:', result.value)
  } else {
    console.log('失败:', result.reason)
  }
})

// Promise.race：第一个完成的
const result = await Promise.race([
  fetchDataWithTimeout(url, 5000),
  timeoutPromise(5000)
])

// Promise.any：第一个成功的（ES2021）
const result = await Promise.any([
  fetchFromServer1(),
  fetchFromServer2(),
  fetchFromServer3()
])

3.2 带超时的 Promise

function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
  const timeout = new Promise<never>((_, reject) => {
    setTimeout(() => reject(new Error(`超时 ${ms}ms`)), ms)
  })
  return Promise.race([promise, timeout])
}

// 使用
const data = await withTimeout(fetchData(url), 5000)

3.3 Promise 链式错误处理

fetchData()
  .then(result => processResult(result))
  .then(output => saveOutput(output))
  .catch(err => {
    // 捕获上面所有阶段的错误
    console.error('流程失败:', err)
    return fallbackValue
  })
  .finally(() => {
    // 无论成功失败都执行
    cleanup()
  })

---

四、并发控制实战

4.1 串行执行

async function serialExecute<T>(tasks: (() => Promise<T>)[]): Promise<T[]> {
  const results: T[] = []
  for (const task of tasks) {
    const result = await task()
    results.push(result)
  }
  return results
}

// 使用
const urls = ['url1', 'url2', 'url3']
const data = await serialExecute(urls.map(url => () => fetch(url)))

4.2 并行执行（无限制）

async function parallelExecute<T>(tasks: (() => Promise<T>)[]): Promise<T[]> {
  return Promise.all(tasks.map(task => task()))
}

// 使用
const data = await parallelExecute(urls.map(url => () => fetch(url)))

4.3 并发数限制（核心）

async function limitExecute<T>(
  tasks: (() => Promise<T>)[],
  limit: number
): Promise<T[]> {
  const results: T[] = []
  const executing = new Set<Promise<T>>()

  for (const task of tasks) {
    const promise = task().then(result => {
      executing.delete(promise)
      return result
    })

    executing.add(promise)
    results.push(promise)

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

  return Promise.all(results)
}

// 使用：最多同时 5 个并发
const data = await limitExecute(urls.map(url => () => fetch(url)), 5)

4.4 异步任务队列

class AsyncQueue<T> {
  private tasks: (() => Promise<T>)[] = []
  private executing = 0
  private maxConcurrency: number
  private resolve?: (() => void)

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

  add(task: () => Promise<T>): Promise<T> {
    return new Promise((resolve, reject) => {
      this.tasks.push(async () => {
        try {
          const result = await task()
          resolve(result)
        } catch (err) {
          reject(err)
        }
      })
      this.run()
    })
  }

  private async run() {
    while (this.executing < this.maxConcurrency && this.tasks.length > 0) {
      const task = this.tasks.shift()!
      this.executing++

      try {
        await task()
      } finally {
        this.executing--
        this.run()
      }
    }

    if (this.executing === 0 && this.tasks.length === 0 && this.resolve) {
      this.resolve()
    }
  }

  done(): Promise<void> {
    if (this.executing === 0 && this.tasks.length === 0) {
      return Promise.resolve()
    }
    return new Promise(resolve => {
      this.resolve = resolve
    })
  }
}

// 使用
const queue = new AsyncQueue(3)
urls.forEach(url => queue.add(() => fetch(url)))
await queue.done()

---

五、限流策略

5.1 时间窗口限流

class RateLimiter {
  private requests: number[] = []
  private maxRequests: number
  private windowMs: number

  constructor(maxRequests: number, windowMs: number) {
    this.maxRequests = maxRequests
    this.windowMs = windowMs
  }

  async acquire(): Promise<void> {
    const now = Date.now()
    
    // 移除窗口外的请求记录
    this.requests = this.requests.filter(t => now - t < this.windowMs)
    
    if (this.requests.length >= this.maxRequests) {
      // 计算需要等待的时间
      const waitTime = this.windowMs - (now - this.requests[0])
      await new Promise(resolve => setTimeout(resolve, waitTime))
      await this.acquire() // 递归检查
    } else {
      this.requests.push(now)
    }
  }
}

// 使用：每秒最多 10 次请求
const limiter = new RateLimiter(10, 1000)
async function fetchWithLimit(url: string) {
  await limiter.acquire()
  return fetch(url)
}

5.2 令牌桶算法

class TokenBucket {
  private tokens: number = 0
  private maxTokens: number
  private refillRate: number // tokens per second
  private lastRefill: number = Date.now()

  constructor(maxTokens: number, refillRate: number) {
    this.maxTokens = maxTokens
    this.refillRate = refillRate
    this.tokens = maxTokens
  }

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

  async acquire(count: number = 1): Promise<void> {
    this.refill()
    
    if (this.tokens >= count) {
      this.tokens -= count
      return
    }

    // 计算需要等待的时间
    const waitTime = ((count - this.tokens) / this.refillRate) * 1000
    await new Promise(resolve => setTimeout(resolve, waitTime))
    await this.acquire(count)
  }
}

// 使用：最大 100 个令牌，每秒补充 10 个
const bucket = new TokenBucket(100, 10)
await bucket.acquire() // 获取 1 个令牌

5.3 漏桶算法

class LeakyBucket {
  private queue: (() => void)[] = []
  private maxSize: number
  private leakRate: number // items per second
  private leaking: boolean = false

  constructor(maxSize: number, leakRate: number) {
    this.maxSize = maxSize
    this.leakRate = leakRate
  }

  async add<T>(task: () => Promise<T>): Promise<T> {
    return new Promise((resolve, reject) => {
      if (this.queue.length >= this.maxSize) {
        reject(new Error('队列已满'))
        return
      }

      this.queue.push(async () => {
        try {
          const result = await task()
          resolve(result)
        } catch (err) {
          reject(err)
        }
      })

      this.startLeaking()
    })
  }

  private startLeaking() {
    if (this.leaking) return
    this.leaking = true

    const leak = () => {
      if (this.queue.length > 0) {
        const task = this.queue.shift()!
        task()
      }

      if (this.queue.length > 0) {
        setTimeout(leak, 1000 / this.leakRate)
      } else {
        this.leaking = false
      }
    }

    leak()
  }
}

// 使用：队列最多 50 个任务，每秒处理 5 个
const bucket = new LeakyBucket(50, 5)
const result = await bucket.add(() => fetch(url))

---

六、重试机制

6.1 简单重试

async function retry<T>(
  task: () => Promise<T>,
  maxAttempts: number,
  delayMs: number = 1000
): Promise<T> {
  let lastError: Error | undefined
  
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      return await task()
    } catch (err) {
      lastError = err as Error
      console.warn(`尝试 ${attempt}/${maxAttempts} 失败:`, err)
      
      if (attempt < maxAttempts) {
        await new Promise(resolve => setTimeout(resolve, delayMs * attempt))
      }
    }
  }
  
  throw lastError || new Error('重试失败')
}

// 使用：最多重试 3 次
const data = await retry(() => fetchData(url), 3)

6.2 指数退避重试

async function retryWithBackoff<T>(
  task: () => Promise<T>,
  maxAttempts: number,
  baseDelay: number = 1000
): Promise<T> {
  let lastError: Error | undefined
  
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      return await task()
    } catch (err) {
      lastError = err as Error
      console.warn(`尝试 ${attempt}/${maxAttempts} 失败:`, err)
      
      if (attempt < maxAttempts) {
        const delay = baseDelay * Math.pow(2, attempt - 1) + Math.random() * 500
        await new Promise(resolve => setTimeout(resolve, delay))
      }
    }
  }
  
  throw lastError || new Error('重试失败')
}

// 使用：指数退避 + 随机抖动
const data = await retryWithBackoff(() => fetchData(url), 5)

6.3 可中断重试

async function retryWithAbort<T>(
  task: () => Promise<T>,
  maxAttempts: number,
  signal?: AbortSignal
): Promise<T> {
  let lastError: Error | undefined
  
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    if (signal?.aborted) {
      throw new Error('操作已取消')
    }
    
    try {
      return await task()
    } catch (err) {
      lastError = err as Error
      
      if (attempt < maxAttempts && !signal?.aborted) {
        await new Promise(resolve => setTimeout(resolve, 1000 * attempt))
      }
    }
  }
  
  throw lastError || new Error('重试失败')
}

// 使用：支持 AbortController 取消
const controller = new AbortController()
const timeout = setTimeout(() => controller.abort(), 30000)

try {
  const data = await retryWithAbort(() => fetchData(url), 5, controller.signal)
} finally {
  clearTimeout(timeout)
}

---

七、Worker Threads 并发

7.1 基本用法

// worker.ts
import { parentPort, workerData } from 'worker_threads'

function fibonacci(n: number): number {
  if (n <= 1) return n
  return fibonacci(n - 1) + fibonacci(n - 2)
}

const result = fibonacci(workerData)
parentPort?.postMessage(result)

// main.ts
import { Worker } from 'worker_threads'

function runWorker(data: number): Promise<number> {
  return new Promise((resolve, reject) => {
    const worker = new Worker('./worker.ts', { workerData: data })
    
    worker.on('message', resolve)
    worker.on('error', reject)
    worker.on('exit', (code) => {
      if (code !== 0) {
        reject(new Error(`Worker exited with code ${code}`))
      }
    })
  })
}

// 使用
const result = await runWorker(40)

7.2 Worker 池

import { Worker, isMainThread, parentPort, workerData } from 'worker_threads'

if (!isMainThread) {
  // Worker 线程
  const result = heavyComputation(workerData)
  parentPort?.postMessage(result)
  process.exit(0)
}

class WorkerPool {
  private workers: Worker[] = []
  private tasks: { resolve: (value: any) => void; data: any }[] = []
  private idleWorkers: Worker[] = []

  constructor(size: number) {
    for (let i = 0; i < size; i++) {
      this.createWorker()
    }
  }

  private createWorker() {
    const worker = new Worker(__filename)
    
    worker.on('message', (result) => {
      const task = this.tasks.shift()
      task?.resolve(result)
      this.idleWorkers.push(worker)
    })

    worker.on('error', () => {
      this.workers = this.workers.filter(w => w !== worker)
      this.createWorker()
    })

    this.workers.push(worker)
    this.idleWorkers.push(worker)
  }

  async execute(data: any): Promise<any> {
    if (this.idleWorkers.length > 0) {
      const worker = this.idleWorkers.pop()!
      worker.postMessage(data)
    }

    return new Promise(resolve => {
      this.tasks.push({ resolve, data })
    })
  }

  destroy() {
    this.workers.forEach(worker => worker.terminate())
  }
}

// 使用
const pool = new WorkerPool(4)
const results = await Promise.all([
  pool.execute(data1),
  pool.execute(data2),
  pool.execute(data3),
  pool.execute(data4)
])
pool.destroy()

---

八、错误处理最佳实践

8.1 全局错误捕获

// 未捕获的 Promise 拒绝
process.on('unhandledRejection', (reason, promise) => {
  console.error('未处理的 Promise 拒绝:', reason)
  // 发送到监控系统
})

// 未捕获的异常
process.on('uncaughtException', (err) => {
  console.error('未捕获的异常:', err)
  // 优雅关闭服务
  server.close(() => process.exit(1))
})

// unhandledRejection 监听后不会自动退出，但 uncaughtException 会

8.2 结构化错误处理

class AppError extends Error {
  constructor(
    public message: string,
    public code: string,
    public status: number = 500,
    public details?: any
  ) {
    super(message)
    this.name = 'AppError'
  }
}

// 使用
async function fetchUser(id: string) {
  const user = await db.query('SELECT * FROM users WHERE id = ?', [id])
  if (!user) {
    throw new AppError('用户不存在', 'USER_NOT_FOUND', 404, { id })
  }
  return user
}

// 中间件统一处理
app.use(async (ctx, next) => {
  try {
    await next()
  } catch (err) {
    if (err instanceof AppError) {
      ctx.status = err.status
      ctx.body = {
        code: err.code,
        message: err.message,
        details: err.details
      }
    } else {
      ctx.status = 500
      ctx.body = { code: 'INTERNAL_ERROR', message: '服务器内部错误' }
    }
  }
})

---

九、性能监控与调试

9.1 异步耗时监控

function traceAsync<T>(name: string, fn: () => Promise<T>): Promise<T> {
  const start = Date.now()
  return fn().then(result => {
    console.log(`${name} 耗时: ${Date.now() - start}ms`)
    return result
  }).catch(err => {
    console.log(`${name} 失败, 耗时: ${Date.now() - start}ms`)
    throw err
  })
}

// 使用
const data = await traceAsync('fetchUser', () => fetchUser(id))

9.2 使用 Clinic.js 分析

# 安装
npm install -g clinic

# CPU 分析
clinic flame -- node app.js

# 内存分析
clinic heap-profiler -- node app.js

# 阻塞分析
clinic bubbleprof -- node app.js

9.3 async_hooks 追踪

import { createHook } from 'async_hooks'

const hook = createHook({
  init(asyncId, type, triggerAsyncId, resource) {
    console.log(`初始化: ${type} (asyncId=${asyncId}, trigger=${triggerAsyncId})`)
  },
  destroy(asyncId) {
    console.log(`销毁: asyncId=${asyncId}`)
  }
})

hook.enable()

---

十、总结

• ✅ 深入理解 Event Loop 6 个阶段及微任务/宏任务
• ✅ 掌握异步编程模式演进（回调 → Promise → async/await）
• ✅ 实战并发控制（串行、并行、并发限制、异步队列）
• ✅ 实现限流策略（时间窗口、令牌桶、漏桶）
• ✅ 实现重试机制（简单重试、指数退避、可中断）
• ✅ Worker Threads 并发编程与 Worker 池
• ✅ 错误处理最佳实践与性能监控

Node.js 的异步编程是一门艺术，掌握这些技巧让你能编写出高效、可靠的代码。

---

相关阅读：

• Node.js 性能调优与内存泄漏排查
• TypeScript 装饰器与元数据编程实战
• Vite 插件开发与构建优化进阶