事件循环阻塞排查

Node.js 单线程事件循环，阻塞操作会导致整个进程停滞，排查阻塞是性能优化核心任务。

事件循环架构

六阶段模型

   ┌───────────────────────────┐
┌─>│           timers          │<─ setTimeout/setInterval
│  └───────────────────────────┘
│  ┌───────────────────────────┐
│  │     pending callbacks     │<─ 上轮未完成的 IO回调
│  └───────────────────────────┘
│  ┌───────────────────────────┐
│  │       idle, prepare       │<─ 内部使用
│  └───────────────────────────┘
│  ┌───────────────────────────┐
│  │           poll            │<─ IO事件、新连接
│  └───────────────────────────┘
│  ┌───────────────────────────┐
│  │           check           │<─ setImmediate
│  └───────────────────────────┘
│  ┌───────────────────────────┐
└─│       close callbacks      │<─ socket.close等
   └───────────────────────────┘

阻塞发生位置

阻塞点：
- timers：超时回调执行时间过长
- poll：IO 回调处理慢
- 任何阶段：同步 CPU 密集操作

阻塞检测方法

monitorEventLoopDelay

const { monitorEventLoopDelay } = require('perf_hooks');

const monitor = monitorEventLoopDelay({
  resolution: 10  // 10ms 粒度
});
monitor.enable();

setInterval(() => {
  console.log({
    mean: monitor.mean.toFixed(2) + 'ms',
    max: monitor.max.toFixed(2) + 'ms',
    p50: monitor.percentile(50).toFixed(2) + 'ms',
    p99: monitor.percentile(99).toFixed(2) + 'ms'
  });
}, 5000);

// 健康指标
// mean < 10ms：健康
// mean > 50ms：阻塞
// max > 100ms：严重阻塞

手动检测延迟

let lastTime = Date.now();

setInterval(() => {
  const now = Date.now();
  const delay = now - lastTime - 1000;  // 应该是1000ms
  if (delay > 50) {
    console.warn(`Event loop lag: ${delay}ms`);
  }
  lastTime = now;
}, 1000);

blocked 库检测

const blocked = require('blocked');

blocked((ms) => {
  console.warn(`Event loop blocked for ${ms}ms`);
}, { threshold: 50 });  // 50ms 以上报警

block-at-first-line 定位

const blockedAt = require('blocked-at');

blockedAt((ms, stack) => {
  console.warn(`Blocked ${ms}ms at:`);
  console.warn(stack);
}, { threshold: 100 });

常见阻塞源

1. 同步 IO 操作

// 阻塞：同步文件读写
app.get('/data', (req, res) => {
  const data = fs.readFileSync('large.json');  // 阻塞！
  res.send(data);
});

// 优化：异步操作
app.get('/data', (req, res) => {
  fs.readFile('large.json', (err, data) => {
    if (err) return res.status(500).send('Error');
    res.send(data);
  });
});

// 优化：流式处理
app.get('/data', (req, res) => {
  fs.createReadStream('large.json').pipe(res);
});

2. CPU 密集计算

// 阻塞：大数组排序
app.post('/sort', (req, res) => {
  const sorted = req.body.data.sort((a, b) => {
    // 大数组排序阻塞
    return a.value - b.value;
  });
  res.send(sorted);
});

// 优化：拆分处理
async function chunkedSort(data, chunkSize = 1000) {
  const chunks = [];
  for (let i = 0; i < data.length; i += chunkSize) {
    chunks.push(data.slice(i, i + chunkSize));
  }

  // 分批排序
  const sortedChunks = chunks.map(chunk => chunk.sort(compare));

  // 合并结果
  return mergeSorted(sortedChunks);
}

// 优化：使用 Worker Threads
const { Worker } = require('worker_threads');

function sortInWorker(data) {
  return new Promise((resolve, reject) => {
    const worker = new Worker('./sort-worker.js', {
      workerData: data
    });
    worker.on('message', resolve);
    worker.on('error', reject);
  });
}

3. 正则表达式回溯

// 阻塞：危险正则（灾难性回溯）
const regex = /^(a+)+$/;
app.post('/validate', (req, res) => {
  const isValid = regex.test(req.body.input);  // 'aaaaaaaaaaaaa!' 阻塞数秒
  res.send({ valid: isValid });
});

// 优化：安全正则
const safeRegex = /^a+$/;  // 无嵌套量词

// 优化：限制输入长度
if (input.length > 100) {
  return res.send({ valid: false });
}

4. JSON.parse 大数据

// 阻塞：解析大 JSON
app.post('/parse', (req, res) => {
  const data = JSON.parse(req.body.json);  // 10MB+ 阻塞
  res.send(data);
});

// 优化：流式解析
const JSONStream = require('JSONStream');

app.post('/parse', (req, res) => {
  req.pipe(JSONStream.parse('*'))
    .on('data', (item) => {
      // 流式处理每个元素
      processItem(item);
    })
    .on('end', () => {
      res.send('done');
    });
});

5. 加密操作

// 阻塞：同步加密
const encrypted = crypto.pbkdf2Sync(password, salt, 100000, 64, 'sha512');

// 优化：异步加密
crypto.pbkdf2(password, salt, 100000, 64, 'sha512', (err, key) => {
  if (err) throw err;
  console.log(key.toString('hex'));
});

// 优化：降低迭代次数（安全范围内）
crypto.pbkdf2(password, salt, 10000, 64, 'sha512', callback);

6. 数据库操作

// 阻塞：同步数据库驱动（部分 ORM）
const result = db.querySync('SELECT * FROM users');  // 阻塞

// 优化：异步查询
db.query('SELECT * FROM users', (err, result) => {
  // 异步回调
});

// 优化：使用 Promise
const result = await db.query('SELECT * FROM users');

阻塞定位工具

clinic.js doctor

# 安装
npm install -g clinic

# 运行诊断
clinic doctor -- node app.js

# 生成报告
# 自动检测事件循环延迟、IO 阻塞

输出解读：

✓ Event loop delay: 5ms      ← 健康
✗ Event loop delay: 150ms    ← 阻塞

✓ CPU usage: 30%             ← 正常
✗ CPU usage: 100%            ← CPU 密集

Potential blocking:
- fs.readFileSync at line 45
- JSON.parse at line 120

Node.js Inspector

# 启动调试
node --inspect app.js

# Chrome DevTools
chrome://inspect

# Performance 面板录制
# 查看函数耗时和调用栈

0x 火焰图定位

npm install -g 0x
0x app.js

# 查看火焰图
# 找到宽度最大的平台（阻塞函数）

阻塞预防策略

拆分长任务

// 问题：一次性处理大量数据
function processAll(items) {
  for (let i = 0; i < items.length; i++) {
    processItem(items[i]);  // 长循环阻塞
  }
}

// 优化：分批处理
async function processBatch(items, batchSize = 100) {
  for (let i = 0; i < items.length; i += batchSize) {
    const batch = items.slice(i, i + batchSize);
    for (const item of batch) {
      processItem(item);
    }
    // 让事件循环有机会处理其他任务
    await new Promise(resolve => setImmediate(resolve));
  }
}

使用 setImmediate 分片

function processLargeArray(array) {
  let index = 0;

  function processChunk() {
    const end = Math.min(index + 100, array.length);
    while (index < end) {
      processItem(array[index++]);
    }

    if (index < array.length) {
      setImmediate(processChunk);  // 让其他任务执行
    } else {
      console.log('Done');
    }
  }

  processChunk();
}

使用 Worker Threads

const { Worker, isMainThread, workerData } = require('worker_threads');

if (isMainThread) {
  // 主线程
  function heavyCompute(data) {
    return new Promise((resolve, reject) => {
      const worker = new Worker(__filename, {
        workerData: data
      });
      worker.on('message', resolve);
      worker.on('error', reject);
      worker.on('exit', (code) => {
        if (code !== 0) reject(new Error(`Worker exited with ${code}`));
      });
    });
  }

  app.post('/compute', async (req, res) => {
    const result = await heavyCompute(req.body);
    res.send(result);
  });
} else {
  // Worker 线程
  const result = compute(workerData);
  parentPort.postMessage(result);
}

限制并发处理

class TaskQueue {
  constructor(concurrency) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = [];
  }

  async run(task) {
    if (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    this.running++;
    try {
      await task();
    } finally {
      this.running--;
      if (this.queue.length) {
        this.queue.shift()();
      }
    }
  }
}

const queue = new TaskQueue(4);  // 最多4个并发

app.post('/process', async (req, res) => {
  await queue.run(() => processData(req.body));
  res.send('done');
});

阻塞排查流程

1. 监控事件循环延迟
   ↓ 延迟 > 50ms
2. 使用 clinic doctor 定位
   ↓ 识别阻塞类型
3. 分析代码热点
   ↓ 火焰图定位函数
4. 应用优化策略
   ↓ 异步/分片/Worker
5. 验证优化效果
   ↓ 延迟降低

监控阈值参考

注意：事件循环阻塞会导致所有请求排队等待，单次阻塞影响整体吞吐量。

要点总结

• 使用 monitorEventLoopDelay API 监控延迟
• 避免 sync IO、CPU 密集操作、危险正则
• 长任务拆分成小块，用 setImmediate 分片
• CPU 密集操作使用 Worker Threads
• 设置并发限制防止资源争抢