异步编程与模式

Node.js 异步编程与模式

Node.js 的核心优势在于异步非阻塞 I/O，但这也带来了回调地狱、错误处理复杂等问题。下面从解救方案、工具函数、并发控制、异步迭代以及常见反模式等角度展开。

一、回调地狱与解救：Promise、async/await、EventEmitter、发布-订阅

1. 回调地狱（Callback Hell）

多层嵌套回调导致代码难以阅读和维护。

fs.readFile("a.txt", (err, dataA) => {
    if (err) throw err;
    fs.readFile("b.txt", (err, dataB) => {
        if (err) throw err;
        fs.readFile("c.txt", (err, dataC) => {
            // 继续嵌套...
        });
    });
});

2. Promise

将异步操作封装为 Promise 链，避免嵌套。

const readFilePromise = file =>
    new Promise((resolve, reject) => {
        fs.readFile(file, (err, data) => (err ? reject(err) : resolve(data)));
    });

readFilePromise("a.txt")
    .then(dataA => readFilePromise("b.txt"))
    .then(dataB => readFilePromise("c.txt"))
    .then(dataC => console.log("done"))
    .catch(err => console.error(err));

3. async/await

基于 Promise 的语法糖，使异步代码像同步一样线性书写。

async function processFiles() {
    try {
        const dataA = await readFilePromise("a.txt");
        const dataB = await readFilePromise("b.txt");
        const dataC = await readFilePromise("c.txt");
        console.log("done");
    } catch (err) {
        console.error(err);
    }
}

4. EventEmitter 与发布-订阅模式

适用于多个异步事件触发多次响应的场景（如流数据、用户交互）。

const EventEmitter = require("events");
const ee = new EventEmitter();

ee.on("data", chunk => {
    console.log("收到数据:", chunk);
});
ee.on("end", () => {
    console.log("完成");
});

// 异步过程
setTimeout(() => {
    ee.emit("data", "hello");
    ee.emit("end");
}, 100);

发布-订阅（如 Redis Pub/Sub）可以跨进程、跨服务器解耦。

二、`util.promisify` 将回调函数转为 Promise

Node.js 内置模块大多采用错误优先回调风格（fn(err, result)）。util.promisify 将这些函数转换为返回 Promise 的版本。

const util = require("util");
const fs = require("fs");

const readFile = util.promisify(fs.readFile);
readFile("file.txt", "utf8")
    .then(data => console.log(data))
    .catch(err => console.error(err));

也适用于自定义函数，只要遵循 (err, res) 签名：

function customApi(arg, callback) {
    // some async work
    callback(null, result);
}
const customPromise = util.promisify(customApi);

三、控制并发

大量异步操作（如批量 API 调用、文件读取）若不加限制，可能耗尽系统资源或触发速率限制。常用库和模式：

1. `p-limit`（轻量级 Promise 并发控制）

const pLimit = require('p-limit');
const limit = pLimit(5); // 最大并发 5

const urls = ['url1', 'url2', ...];
const requests = urls.map(url => limit(() => fetch(url)));
const results = await Promise.all(requests);

2. `async.queue`（基于回调的队列）

const { queue } = require("async");
const q = queue(async task => {
    await process(task);
}, 5); // 并发数

q.push(tasks);
q.drain(() => console.log("全部完成"));

3. 自定义令牌桶（Token Bucket）实现更精细限流

class TokenBucket {
    constructor(capacity, fillPerSecond) {
        this.capacity = capacity;
        this.tokens = capacity;
        setInterval(() => {
            this.tokens = Math.min(this.capacity, this.tokens + fillPerSecond);
        }, 1000);
    }
    async take() {
        while (this.tokens <= 0) await delay(10);
        this.tokens--;
    }
}

四、异步迭代器与 `for await...of` 处理流式数据

Node.js 可读流（Readable）实现了异步迭代器，可以用 for await...of 逐块读取数据。

const { createReadStream } = require("fs");
const stream = createReadStream("large.txt", { encoding: "utf8" });

(async () => {
    for await (const chunk of stream) {
        console.log("收到 chunk:", chunk.length);
    }
    console.log("流结束");
})();

自定义异步可迭代对象：

const asyncIterable = {
    [Symbol.asyncIterator]() {
        let i = 0;
        return {
            next() {
                if (i < 3) return Promise.resolve({ value: i++, done: false });
                return Promise.resolve({ done: true });
            },
        };
    },
};
for await (const val of asyncIterable) console.log(val);

在数据库游标、HTTP 分块响应等场景非常有用。

五、常见反模式

1. 未捕获的异步错误

没有 .catch 的 Promise 或未 try/catch 的 await 会导致静默失败或进程崩溃。
解决方案：始终添加捕获，或使用 process.on('unhandledRejection', handler)。

2. 无限制并发

同时发起成千上万个异步操作（如 Promise.all 大量请求）可能耗尽内存、文件描述符或触发服务端限流。
解决方案：使用上述并发控制库或限流器。

3. 同步方法混用（阻塞事件循环）

在回调或 Promise 执行体中调用 fs.readFileSync、while 循环或 CPU 密集型同步代码。
解决方案：使用异步版本或将其剥离到 worker_threads。

4. 回调地狱的残余：过度使用 `Promise.then` 链而非 `async/await`

虽然 Promise 解决了嵌套，但长链仍难读。
解决方案：优先使用 async/await。

5. 忘记 `await` 关键字

async function test() {
    const result = fetchData(); // 少了 await，result 是 Promise，不是实际值
}

解决方案：启用 ESLint 规则 @typescript-eslint/await-thenable 或 require-await。

6. 事件监听器中的错误未捕获

若 EventEmitter 触发 error 事件但没有监听，Node.js 会将其视为未捕获异常并可能崩溃。
解决方案：始终监听 error 事件。

总结

模式/工具	作用
Promise + async/await	避免回调地狱，线性编写异步代码
`util.promisify`	将旧回调函数转为 Promise
并发控制（`p-limit`、`async.queue`、令牌桶）	限制同时进行的异步操作数量
异步迭代器 + `for await...of`	流式处理数据块
反模式预防	捕获错误、控制并发、避免同步阻塞

掌握这些模式，可以写出健壮、高性能的 Node.js 异步代码。