【Node.js】事件循环深度解析：从 libuv 到"两层轮回"的异步密码

1
// 浏览器中输出：Promise > setTimeout
2
Promise.resolve().then(() => console.log('Promise'))
3
setTimeout(() => console.log('setTimeout'), 0)

1
// Node.js中输出：process.nextTick > Promise > setTimeout
2
process.nextTick(() => console.log('process.nextTick'))
3
Promise.resolve().then(() => console.log('Promise'))
4
setTimeout(() => console.log('setTimeout'), 0)

1
do {
2
  // 1. 运行小乘轮回（处理libuv的事件）
3
  int run_result = uv_run(env->event_loop(), UV_RUN_DEFAULT);
4
  if (run_result != 0) break; // 处理错误
5

6
  // 2. 处理V8的任务（如Promise微任务、setTimeout回调）
7
  platform->DrainTasks(isolate);
8

9
  // 3. 检查是否还有活跃事件（如未完成的I/O、新的定时器）
10
  bool more_events = uv_loop_alive(env->event_loop()) != 0;
11

12
  // 4. 检查是否需要停止（如process.exit()）
13
} while (more_events && !env->is_stopping());

1
while (loop->alive && !loop->stop) {
2
  uv__update_time(loop);       // 1. 更新当前时间（用于定时器）
3
  uv__run_timers(loop);        // 2. 处理定时器（setTimeout/setInterval）
4
  uv__run_pending(loop);       // 3. 处理延迟的I/O事件（如TCP连接失败）
5
  uv__run_idle(loop);          // 4. 处理Idle事件（setImmediate）
6
  uv__run_prepare(loop);       // 5. 准备I/O监听（告知V8要进入阻塞）
7
  uv__io_poll(loop, timeout);  // 6. 等待I/O事件（阻塞，直到超时或有事件）
8
  uv__run_check(loop);         // 7. 处理Check事件（I/O后的复查）
9
  uv__run_closing_handles(loop);// 8. 处理关闭的句柄（如socket.close()）
10
}

1
// 模拟长任务：占用主线程1.5秒
2
function longTask() {
3
  let sum = 0
4
  for (let i = 0; i < 1e8; i++) sum += i
5
}
6

7
// 1秒后执行的定时器
8
setTimeout(() => console.log('Timer executed'), 1000)
9

10
// 执行长任务，阻塞事件循环
11
longTask()

1
const net = require('net')
2

3
// 连接一个不存在的端口
4
const client = net.connect({ port: 8080 })
5

6
client.on('error', (err) => console.log('Error:', err.message))
7
setTimeout(() => console.log('Timer'), 0)

1
const fs = require('fs')
2

3
// 读取文件的回调中，setImmediate比setTimeout(0)先执行
4
fs.readFile(__filename, () => {
5
  setImmediate(() => console.log('setImmediate'))
6
  setTimeout(() => console.log('setTimeout'), 0)
7
})

1
const fs = require('fs')
2

3
// 读取大文件（1GB）
4
fs.readFile('large-file.txt', (err, data) => {
5
  console.log('File read done')
6
})
7

8
// 1秒后执行的定时器
9
setTimeout(() => console.log('Timer'), 1000)

1
// main.js：主线程
2
const { Worker } = require('worker_threads')
3

4
// 启动worker处理统计任务
5
function startStatistics() {
6
  const worker = new Worker('./statistics.js')
7

8
  worker.on('message', (result) => {
9
    console.log('统计结果：', result)
10
  })
11

12
  worker.on('error', (err) => {
13
    console.error('统计失败：', err)
14
  })
15
}
16

17
// 启动HTTP服务，处理订单请求
18
const http = require('http')
19
const server = http.createServer((req, res) => {
20
  // 模拟订单处理
21
  res.end('订单已接收')
22
  startStatistics() // 启动统计任务
23
})
24

25
server.listen(3000)

1
// statistics.js：worker线程
2
function calculateSales() {
3
  let sum = 0
4
  // 模拟统计100万笔订单
5
  for (let i = 0; i < 1e6; i++) sum += i
6
  return sum
7
}
8

9
// 向主线程发送结果
10
parentPort.postMessage({
11
  totalSales: calculateSales(),
12
})

1
const fs = require('fs')
2
const redis = require('redis')
3
const client = redis.createClient()
4

5
// 读取库存文件后，用setImmediate更新Redis
6
fs.readFile('stock.json', (err, data) => {
7
  if (err) throw err
8
  const stock = JSON.parse(data)
9

10
  // 用setImmediate确保在I/O之后执行
11
  setImmediate(() => {
12
    client.set('stock:iphone', stock.iphone, (err) => {
13
      if (err) throw err
14
      console.log('库存更新完成')
15
    })
16
  })
17
})

1
const mysql = require('mysql')
2

3
const connection = mysql.createConnection({
4
  host: 'localhost',
5
  user: 'root',
6
  password: 'password',
7
})
8

9
connection.connect((err) => {
10
  if (err) {
11
    // 紧急任务：记录错误并退出
12
    process.nextTick(() => {
13
      console.error('数据库连接失败：', err)
14
      process.exit(1)
15
    })
16
  }
17
})

1
<template>
2
  <div class="order-form">
3
    <input
4
      v-model="address"
5
      placeholder="收货地址"
6
    />
7
    <button
8
      @click="submitOrder"
9
      :disabled="isSubmitting"
10
    >
11
      提交订单
12
    </button>
13
  </div>
14
</template>
15

16
<script setup>
17
import { ref } from 'vue'
18
import axios from 'axios'
19

20
const address = ref('')
21
const isSubmitting = ref(false)
22

23
async function submitOrder() {
24
  if (!address.value) return alert('请填写收货地址')
25
  isSubmitting.value = true
26
  try {
27
    const response = await axios.post('/api/orders', {
28
      userId: 123,
29
      address: address.value,
30
      items: [{ productId: 456, quantity: 2 }],
31
    })
32
    if (response.data.success) {
33
      alert('订单提交成功！')
34
      address.value = ''
35
    } else {
36
      alert('订单提交失败：' + response.data.message)
37
    }
38
  } catch (err) {
39
    console.error('提交订单错误：', err)
40
    alert('服务器错误，请重试')
41
  } finally {
42
    isSubmitting.value = false
43
  }
44
}
45
</script>

1
const express = require('express')
2
const app = express()
3
const { Order, Product } = require('./models')
4
const fs = require('fs').promises
5

6
app.use(express.json())
7

8
// 提交订单接口
9
app.post('/api/orders', async (req, res) => {
10
  const { userId, address, items } = req.body
11
  try {
12
    // 1. 创建订单（数据库I/O，快速完成）
13
    const order = await Order.create({ userId, address, items })
14
    // 2. 用setImmediate处理后续耗时任务（扣减库存、发送通知）
15
    setImmediate(async () => {
16
      // 扣减库存
17
      for (const item of items) {
18
        await Product.decrement('stock', {
19
          by: item.quantity,
20
          where: { id: item.productId },
21
        })
22
      }
23
      // 发送短信通知（调用第三方API）
24
      await axios.post('https://sms-api.com/send', {
25
        phone: '138xxxx1234',
26
        content: `您的订单${order.id}已提交，等待发货`,
27
      })
28
      console.log(`📦 订单${order.id}后续任务处理完成`)
29
    })
30
    // 3. 立即返回成功响应，提升用户体验
31
    res.json({ success: true, orderId: order.id })
32
  } catch (err) {
33
    console.error('提交订单错误：', err)
34
    res.status(500).json({ success: false, message: '服务器错误' })
35
  }
36
})
37

38
app.listen(3000, () => console.log('后端服务启动：http://localhost:3000'))