事件循环调度和执行异步任务机制
- 事件循环基础:Node.js 的事件循环是一个持续运行的机制,它不断检查事件队列中是否有任务。事件循环分为多个阶段,如
timers
、I/O callbacks
、idle, prepare
、poll
、check
、close callbacks
等。
- 异步任务调度:
- 宏任务(macrotask):包括
setTimeout
、setInterval
、setImmediate
、I/O
操作、script(整体代码)
等。事件循环每次从宏任务队列中取出一个宏任务执行,执行完后进入下一个阶段。
- 微任务(microtask):如
Promise.then
、process.nextTick
等。在每个宏任务执行完后,事件循环会清空微任务队列,即不断从微任务队列中取出任务执行,直到微任务队列为空,才进入下一个宏任务或事件循环的下一阶段。
- 示例代码:
console.log('script start');
setTimeout(() => {
console.log('setTimeout');
}, 0);
Promise.resolve()
.then(() => {
console.log('Promise.then');
});
console.log('script end');
- 在上述代码中,
script start
和 script end
属于整体代码(宏任务),先执行。Promise.then
是微任务,在宏任务执行完后,清空微任务队列时执行。setTimeout
是宏任务,在本次宏任务执行完,微任务队列清空后,下次事件循环从宏任务队列中取出执行。
高并发场景下避免异步操作过多性能问题的解决方案
- 限制并发数量:
- 原理:通过控制同时执行的异步任务数量,避免过多任务同时竞争资源导致性能下降。
- 使用
Promise.allSettled
和队列:
function asyncTask(id) {
return new Promise((resolve) => {
setTimeout(() => {
console.log(`Task ${id} completed`);
resolve();
}, Math.floor(Math.random() * 1000));
});
}
function limitConcurrentTasks(tasks, limit) {
return new Promise((resolve) => {
let completed = 0;
const results = [];
const queue = tasks.slice();
const runTask = () => {
if (queue.length === 0 && completed === tasks.length) {
resolve(results);
return;
}
while (queue.length > 0 && Promise.allSettled(results).length < limit) {
const task = queue.shift();
task().then((result) => {
results.push({ status: 'fulfilled', value: result });
}).catch((error) => {
results.push({ status:'rejected', reason: error });
}).finally(() => {
completed++;
runTask();
});
}
};
runTask();
});
}
const tasks = Array.from({ length: 10 }, (_, i) => () => asyncTask(i + 1));
limitConcurrentTasks(tasks, 3).then(() => {
console.log('All tasks completed');
});
- 在上述代码中,
limitConcurrentTasks
函数实现了限制并发任务数量为 limit
。通过维护一个任务队列 queue
,每次从队列中取出任务执行,当执行中的任务数量小于 limit
时,继续从队列中取任务执行,直到任务全部完成。
- 使用
cluster
模块(适用于多核 CPU 场景):
- 原理:
cluster
模块允许 Node.js 应用程序创建多个工作进程,充分利用多核 CPU 的优势,将负载分布到不同的进程中。
- 示例代码:
const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;
if (cluster.isMaster) {
console.log(`Master ${process.pid} is running`);
for (let i = 0; i < numCPUs; i++) {
cluster.fork();
}
cluster.on('exit', (worker, code, signal) => {
console.log(`worker ${worker.process.pid} died`);
});
} else {
http.createServer((req, res) => {
res.writeHead(200);
res.end('Hello World\n');
}).listen(8000, () => {
console.log(`Worker ${process.pid} started`);
});
}
- 在上述代码中,主进程(
isMaster
)创建多个工作进程(cluster.fork()
),每个工作进程都监听相同的端口(这里是 8000),实现负载均衡,提高应用程序在高并发场景下的性能。