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 插件開發與構建優化進階