异步性能优化技巧

异步性能优化技巧

Node.js 的异步非阻塞 I/O 模型是其高性能的核心，但不当的异步代码编写反而可能导致性能下降。合理利用异步特性需要掌握事件循环机制、任务调度策略和资源管理技巧。

理解事件循环阶段

Node.js 事件循环分为六个主要阶段：

1. Timers：执行 setTimeout 和 setInterval 回调
2. Pending callbacks：执行系统操作回调（如 TCP 错误）
3. Idle/Prepare：内部使用
4. Poll：检索新 I/O 事件，执行相关回调
5. Check：执行 setImmediate 回调
6. Close callbacks：执行关闭事件回调（如 socket.on('close')）

// 演示阶段执行顺序
setImmediate(() => console.log('Check阶段'));
setTimeout(() => console.log('Timers阶段'), 0);
fs.readFile(__filename, () => {
  console.log('Poll阶段');
  setImmediate(() => console.log('嵌套Check阶段'));
});

任务队列优先级控制

微任务（Promise、process.nextTick）具有最高优先级，会插队执行：

Promise.resolve().then(() => console.log('微任务1'));
process.nextTick(() => console.log('nextTick'));
Promise.resolve().then(() => console.log('微任务2'));
// 输出顺序：nextTick → 微任务1 → 微任务2

优化建议：

• 避免在热点路径中过度使用 process.nextTick
• 耗时同步操作应包装为 setImmediate 释放事件循环

流式处理大数据集

传统方式会导致内存暴涨：

// 反模式
fs.readFile('huge.log', (err, data) => {
  const lines = data.toString().split('\n'); // 内存峰值
  processLines(lines);
});

改用流处理：

fs.createReadStream('huge.log')
  .pipe(split2()) // 按行分割
  .on('data', line => processLine(line))
  .on('end', () => console.log('处理完成'));

集群模式利用多核CPU

单线程Node.js实例无法充分利用多核：

const cluster = require('cluster');
const os = require('os');

if (cluster.isMaster) {
  // 根据CPU核心数创建子进程
  os.cpus().forEach(() => cluster.fork()); 
} else {
  require('./app'); // 每个子进程运行应用实例
}

异步并发控制

无限制并发会导致资源耗尽：

// 危险示范
urls.forEach(url => fetch(url).then(processResponse));

使用 p-limit 控制并发：

const limit = require('p-limit');
const concurrency = 10;
const limiter = limit(concurrency);

Promise.all(
  urls.map(url => 
    limiter(() => fetch(url).then(processResponse))
  )
);

内存泄漏预防

异步代码常见泄漏场景：

// 闭包保持大对象引用
server.on('request', (req) => {
  const hugeObj = loadHugeData();
  req.on('close', () => {
    // hugeObj 无法释放
  });
});

解决方案：

• 使用 WeakMap 替代强引用
• 显式清理事件监听器

定时器优化技巧

避免定时器堆积：

// 错误方式
function pollUpdates() {
  fetchUpdates().then(() => {
    setTimeout(pollUpdates, 100); // 可能重叠执行
  });
}

改进方案：

let isUpdating = false;
async function pollUpdates() {
  if (isUpdating) return;
  isUpdating = true;
  await fetchUpdates();
  isUpdating = false;
  setTimeout(pollUpdates, 100);
}

异步堆栈追踪

默认异步错误丢失调用栈：

function foo() {
  setTimeout(() => {
    throw new Error('难追踪的错误');
  }, 100);
}

启用 async_hooks 增强追踪：

const async_hooks = require('async_hooks');
const fs = require('fs');

async_hooks.createHook({
  init(asyncId, type, triggerAsyncId) {
    fs.writeSync(1, `${type}(${asyncId}): trigger: ${triggerAsyncId}\n`);
  }
}).enable();

Promise 性能陷阱

链式 Promise 创建过多微任务：

// 低效写法
users.reduce((promise, user) => {
  return promise.then(() => updateUser(user));
}, Promise.resolve());

批量处理优化：

const BATCH_SIZE = 100;
for (let i = 0; i < users.length; i += BATCH_SIZE) {
  await Promise.all(
    users.slice(i, i+BATCH_SIZE).map(updateUser)
  );
}

异步资源管理

未释放的资源会导致文件描述符泄漏：

// 忘记关闭的数据库连接
const db = await connectDB();
app.get('/', async () => {
  return db.query('SELECT...'); 
});

使用 AsyncDisposable (Node.js 20+)：

const { open } = require('node:fs/promises');

async function processFile() {
  await using file = await open('data.txt');
  // 自动关闭文件
}

Worker Threads 优化 CPU 密集型任务

将计算转移到工作线程：

const { Worker } = require('worker_threads');

function runInWorker(script, data) {
  return new Promise((resolve) => {
    const worker = new Worker(script, { workerData: data });
    worker.on('message', resolve);
  });
}

// worker.js
const { workerData, parentPort } = require('worker_threads');
parentPort.postMessage(heavyCompute(workerData));

事件发射器优化

高频事件需要特别处理：

const { EventEmitter } = require('events');
class Sensor extends EventEmitter {
  constructor() {
    super();
    this.setMaxListeners(100); // 避免内存泄漏警告
    this.cache = new Map();
  }
  
  onData(data) {
    if (!this.cache.has(data.id)) {
      this.emit('new-data', data); // 去重触发
      this.cache.set(data.id, Date.now());
    }
  }
}

异步初始化模式

应用启动时并行初始化：

async function initApp() {
  const [db, config, cache] = await Promise.all([
    connectDatabase(),
    loadConfig(),
    initCache()
  ]);
  
  return { db, config, cache };
}

实时应用优化

WebSocket 消息批处理：

const messageQueue = [];
let isProcessing = false;

socket.on('message', (msg) => {
  messageQueue.push(msg);
  if (!isProcessing) {
    isProcessing = true;
    setImmediate(processQueue);
  }
});

function processQueue() {
  const batch = messageQueue.splice(0, 100);
  if (batch.length) {
    saveToDB(batch).finally(() => {
      if (messageQueue.length) setImmediate(processQueue);
      else isProcessing = false;
    });
  }
}

错误处理策略

未捕获的 Promise 会导致静默失败：

// 全局Promise错误捕获
process.on('unhandledRejection', (err) => {
  metrics.increment('unhandled_rejection');
  logger.error('Unhandled rejection', err);
});

// 上下文保留的错误包装
async function wrapAsync(fn) {
  return async (...args) => {
    try {
      return await fn(...args);
    } catch (err) {
      err.context = { args };
      throw err;
    }
  };
}

性能分析工具

使用 async_hooks 监控异步延迟：

const hooks = require('async_hooks');
const active = new Map();

hooks.createHook({
  init(id, type, triggerId) {
    if (type === 'Timeout') {
      active.set(id, {
        start: process.hrtime.bigint(),
        triggerId
      });
    }
  },
  destroy(id) {
    const record = active.get(id);
    if (record) {
      const duration = Number(process.hrtime.bigint() - record.start) / 1e6;
      console.log(`Timeout took ${duration}ms`);
      active.delete(id);
    }
  }
}).enable();