事件循环

┌─────────────────────────────────────────────────────┐
│                    Node.js 应用                      │
└─────────────────────┬───────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────┐
│                  V8 JavaScript 引擎                  │
└─────────────────────┬───────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────┐
│                     libuv                            │
│  ┌──────────────┐  ┌──────────────┐                │
│  │  事件循环     │  │  线程池       │                │
│  │  (Event Loop)│  │(Thread Pool) │                │
│  └──────────────┘  └──────────────┘                │
│  ┌──────────────┐  ┌──────────────┐                │
│  │  文件系统     │  │  网络 I/O    │                │
│  └──────────────┘  └──────────────┘                │
└─────────────────────────────────────────────────────┘

   ┌───────────────────────┐
┌─>│        timers         │ setTimeout/setInterval
│  └───────────┬───────────┘
│  ┌───────────┴───────────┐
│  │   pending callbacks   │ 执行延迟到下一轮循环的 I/O 回调
│  └───────────┬───────────┘
│  ┌───────────┴───────────┐
│  │       idle, prepare   │ libuv 内部使用
│  └───────────┬───────────┘
│  ┌───────────┴───────────┐
│  │         poll         │ 轮询新的 I/O 事件
│  └───────────┬───────────┘
│  ┌───────────┴───────────┐
│  │        check         │ setImmediate
│  └───────────┬───────────┘
│  ┌───────────┴───────────┐
└──┤    close callbacks    │ socket.on('close')
   └───────────────────────┘

setTimeout(() => {
  console.log("定时器到期")
}, 100)

const socket = net.connect({ port: 9999 })
socket.on("error", (err) => {
  // 这个回调可能在此阶段执行
})

const fs = require("fs")

fs.readFile("data.txt", (err, data) => {
  // 这个回调在 poll 阶段执行
  console.log("文件读取完成")
})

setImmediate(() => {
  console.log("立即执行")
})

socket.on("close", () => {
  console.log("连接关闭")
})

process.nextTick(() => {
  console.log("nextTick")
})

Promise.resolve().then(() => {
  console.log("Promise")
})

console.log("同步")

// 输出：
// 同步
// nextTick
// Promise

Promise.resolve().then(() => {
  console.log("Promise 微任务")
})

async function asyncDemo() {
  await Promise.resolve()
  console.log("await 后的代码")
}

// 顺序不确定
setTimeout(() => console.log("timeout"), 0)
setImmediate(() => console.log("immediate"))

const fs = require("fs")

fs.readFile(__filename, () => {
  setTimeout(() => console.log("timeout"), 0)
  setImmediate(() => console.log("immediate"))
})

// 输出：
// immediate
// timeout

const fs = require("fs")
const cache = new Map()

function readFileCached(filename, callback) {
  if (cache.has(filename)) {
    // 使用 nextTick 确保回调始终异步执行
    process.nextTick(() => callback(null, cache.get(filename)))
  } else {
    fs.readFile(filename, callback)
  }
}

function promisifyCallback(fn) {
  return function(...args) {
    return new Promise((resolve, reject) => {
      try {
        fn.call(this, ...args, (err, result) => {
          if (err) reject(err)
          else resolve(result)
        })
      } catch (err) {
        // 使用 nextTick 确保错误正确传播
        process.nextTick(() => reject(err))
      }
    })
  }
}

// ❌ 递归调用会阻塞事件循环
function bad() {
  process.nextTick(bad)
}

// ✅ 使用 setImmediate 让 I/O 有机会执行
function good() {
  setImmediate(good)
}

console.log("1. 同步开始")

setTimeout(() => {
  console.log("2. setTimeout")
}, 0)

setImmediate(() => {
  console.log("3. setImmediate")
})

process.nextTick(() => {
  console.log("4. nextTick")
})

Promise.resolve().then(() => {
  console.log("5. Promise")
})

console.log("6. 同步结束")

// 输出顺序：
// 1. 同步开始
// 6. 同步结束
// 4. nextTick
// 5. Promise
// 2 或 3（不确定）

const fs = require("fs")

fs.readFile(__filename, () => {
  console.log("1. readFile")

  setTimeout(() => {
    console.log("2. setTimeout")
  }, 0)

  setImmediate(() => {
    console.log("3. setImmediate")
  })

  process.nextTick(() => {
    console.log("4. nextTick")
  })
})

console.log("0. 同步")

// 输出顺序：
// 0. 同步
// 1. readFile
// 4. nextTick
// 3. setImmediate
// 2. setTimeout

function heavyCompute() {
  let sum = 0
  for (let i = 0; i < 1e9; i++) {
    sum += i
  }
  return sum
}

console.time("compute")
heavyCompute() // 阻塞事件循环
console.timeEnd("compute")

async function heavyComputeChunked(total, chunkSize = 1e6) {
  let sum = 0
  
  for (let i = 0; i < total; i += chunkSize) {
    const end = Math.min(i + chunkSize, total)
    for (let j = i; j < end; j++) {
      sum += j
    }
    // 让出控制权
    await new Promise(resolve => setImmediate(resolve))
  }
  
  return sum
}

const { Worker } = require("worker_threads")

function runInWorker(data) {
  return new Promise((resolve, reject) => {
    const worker = new Worker("./compute.js", { workerData: data })
    worker.on("message", resolve)
    worker.on("error", reject)
  })
}

const result = await runInWorker({ n: 1e9 })

// 查看线程池大小
console.log(process.env.UV_THREADPOOL_SIZE) // 默认 4

// 设置线程池大小（启动前设置）
process.env.UV_THREADPOOL_SIZE = 8

const { monitorEventLoopDelay } = require("perf_hooks")

const histogram = monitorEventLoopDelay({ resolution: 20 })
histogram.enable()

setInterval(() => {
  console.log("事件循环延迟:", histogram.mean / 1e6, "ms")
}, 1000)

const async_hooks = require("async_hooks")

const hook = async_hooks.createHook({
  init(asyncId, type, triggerAsyncId) {
    console.log("init", { asyncId, type, triggerAsyncId })
  },
  destroy(asyncId) {
    console.log("destroy", asyncId)
  }
})

hook.enable()

// ❌ CPU 密集型任务阻塞事件循环
function bad() {
  for (let i = 0; i < 1e9; i++) {
    // 长时间计算
  }
}

// ✅ 分块处理或使用 Worker Threads
async function good() {
  // 分块处理
  for (let i = 0; i < 1e9; i += 1e6) {
    // 处理一块
    await new Promise(resolve => setImmediate(resolve))
  }
}

// ❌ 过度使用会导致 I/O 饥饿
process.nextTick(function recursive() {
  process.nextTick(recursive)
})

// ✅ 使用 setImmediate
setImmediate(function recursive() {
  setImmediate(recursive)
})

// 需要 I/O 后立即执行 → setImmediate
fs.readFile("data.txt", () => {
  setImmediate(() => {
    // 在下一个 check 阶段执行
  })
})

// 需要最小延迟 → setTimeout
setTimeout(() => {
  // 在 timers 阶段执行
}, 0)

// 让出主线程，避免长任务阻塞事件循环
async function processItems(items) {
  for (const item of items) {
    await new Promise(resolve => setImmediate(resolve))  // 让出控制权
    processItem(item)
  }
}

import { performance, eventLoopUtilization } from 'node:perf_hooks'

// 事件循环利用率
const elu = eventLoopUtilization()
console.log(`事件循环利用率: ${(elu.utilization * 100).toFixed(2)}%`)

// 性能观察者
const obs = new PerformanceObserver((list) => {
  for (const entry of list.getEntries()) {
    console.log(`${entry.name}: ${entry.duration}ms`)
  }
})
obs.observe({ entryTypes: ['measure', 'function'] })

// 测量异步函数执行时间
performance.mark('start')
await someAsyncOperation()
performance.mark('end')
performance.measure('operation', 'start', 'end')