Node.js事件循环机制深度解析

// setTimeout的执行时机取决于事件循环当前所处阶段
const start = Date.now();

setTimeout(() => {
    console.log(`Timer executed after ${Date.now() - start}ms`);
}, 100);

// 模拟一个耗时操作
const data = require('fs').readFileSync('/dev/urandom', { length: 100000 });
// readFileSync是同步的，会阻塞事件循环
// 如果这里耗时200ms，那么setTimeout回调要等到200ms后才执行

// process.nextTick vs Promise vs queueMicrotask
// process.nextTick优先级最高

Promise.resolve().then(() => console.log('1: Promise'));
queueMicrotask(() => console.log('2: queueMicrotask'));
process.nextTick(() => console.log('3: nextTick'));

// 输出顺序:
// 3: nextTick
// 1: Promise         (Promise和queueMicrotask在同一队列)
// 2: queueMicrotask

// 完整的执行顺序示例
console.log('1: script start');

setTimeout(() => {
    console.log('2: setTimeout');
    Promise.resolve().then(() => {
        console.log('3: setTimeout > Promise');
    });
}, 0);

setImmediate(() => {
    console.log('4: setImmediate');
    process.nextTick(() => {
        console.log('5: setImmediate > nextTick');
    });
});

Promise.resolve().then(() => {
    console.log('6: Promise');
});

process.nextTick(() => {
    console.log('7: nextTick');
    process.nextTick(() => {
        console.log('8: nextTick > nextTick');
    });
});

console.log('9: script end');

// 输出:
// 1: script start
// 9: script end
// 7: nextTick
// 8: nextTick > nextTick
// 6: Promise
// 2: setTimeout      (Timer和Immediate的顺序在主模块中不确定)
// 3: setTimeout > Promise
// 4: setImmediate
// 5: setImmediate > nextTick

// 在主模块中，顺序不确定（取决于系统性能和事件循环启动时间）
setTimeout(() => console.log('timeout'), 0);
setImmediate(() => console.log('immediate'));
// 可能是 timeout -> immediate，也可能是 immediate -> timeout

// 但在I/O回调中，setImmediate总是先执行
const fs = require('fs');
fs.readFile(__filename, () => {
    setTimeout(() => console.log('timeout'), 0);
    setImmediate(() => console.log('immediate'));
});
// 总是: immediate -> timeout
// 因为I/O回调在Poll阶段执行，下一个阶段是Check（setImmediate）

// process.nextTick递归会阻塞事件循环（I/O饥饿）
// 不推荐的写法
function recursiveNextTick() {
    process.nextTick(recursiveNextTick);
    // 永远不会进入下一个事件循环阶段！
}

// 推荐使用setImmediate替代
function recursiveImmediate() {
    setImmediate(recursiveImmediate);
    // 每次在Check阶段执行，不会阻塞其他阶段
}

// process.nextTick的合理使用场景
class EventEmitterExample extends require('events') {
    constructor() {
        super();
        // 确保事件处理器在构造函数返回后才触发
        process.nextTick(() => {
            this.emit('ready');
        });
    }
}

// 用户可以在构造后注册事件处理器
const emitter = new EventEmitterExample();
emitter.on('ready', () => {
    console.log('ready event fired');
});

// 调整线程池大小
// 在程序启动前设置环境变量
// UV_THREADPOOL_SIZE=16 node app.js

// 或在代码中设置（必须在require任何模块之前）
process.env.UV_THREADPOOL_SIZE = 16;

// 线程池满时的排队行为
const crypto = require('crypto');
const start = Date.now();

// 默认4个线程，5个并发pbkdf2
for (let i = 0; i < 5; i++) {
    crypto.pbkdf2('password', 'salt', 100000, 512, 'sha512', () => {
        console.log(`pbkdf2 ${i}: ${Date.now() - start}ms`);
    });
}

// 输出示例（4线程）:
// pbkdf2 0: 850ms  \
// pbkdf2 1: 855ms   > 前4个几乎同时完成
// pbkdf2 2: 860ms  /
// pbkdf2 3: 862ms /
// pbkdf2 4: 1710ms  -> 第5个等待线程释放，耗时约2倍

// 不良实践：同步操作阻塞事件循环
const data = JSON.parse(fs.readFileSync('huge-file.json', 'utf8'));

// 良好实践：使用流或异步操作
const { pipeline } = require('stream/promises');
const { createReadStream } = require('fs');
const JSONStream = require('JSONStream');

async function processLargeJSON(filePath) {
    const results = [];
    const readStream = createReadStream(filePath);
    const parser = JSONStream.parse('*');

    parser.on('data', (item) => {
        results.push(item);
    });

    await pipeline(readStream, parser);
    return results;
}

// 使用perf_hooks监控事件循环延迟
const { monitorEventLoopDelay } = require('perf_hooks');

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

setInterval(() => {
    console.log({
        min: histogram.min / 1e6,     // 转换为毫秒
        max: histogram.max / 1e6,
        mean: histogram.mean / 1e6,
        p99: histogram.percentile(99) / 1e6,
    });
    histogram.reset();
}, 5000);

// 简单的事件循环延迟检测
function monitorLoop(threshold = 100) {
    let lastCheck = Date.now();

    setInterval(() => {
        const now = Date.now();
        const delay = now - lastCheck - 1000; // 预期间隔1000ms
        if (delay > threshold) {
            console.warn(`Event loop delay: ${delay}ms`);
        }
        lastCheck = now;
    }, 1000).unref(); // unref防止阻止进程退出
}

// main.js
const { Worker, isMainThread, parentPort } = require('worker_threads');

if (isMainThread) {
    function runWorker(data) {
        return new Promise((resolve, reject) => {
            const worker = new Worker(__filename, {
                workerData: data
            });
            worker.on('message', resolve);
            worker.on('error', reject);
        });
    }

    async function main() {
        const results = await Promise.all([
            runWorker({ task: 'hash', data: 'password1' }),
            runWorker({ task: 'hash', data: 'password2' }),
        ]);
        console.log(results);
    }

    main();
} else {
    const { workerData } = require('worker_threads');
    const crypto = require('crypto');

    const hash = crypto.pbkdf2Sync(
        workerData.data, 'salt', 100000, 64, 'sha512'
    );
    parentPort.postMessage(hash.toString('hex'));
}