如何通过队列缓存数据块，用async/await处理Node.js流背压？

你的实现思路方向是对的，但存在几个关键问题，可能导致并发写入时的逻辑混乱、数据丢失或Promise挂起：

现有实现的问题 #

• 并发竞态风险：循环中直接调用write且不await，多个write会同时执行，可能导致队列被多个函数实例同时修改，比如多个实例同时进入队列处理循环，引发重复写入或队列操作错误。
• 状态判断时机失效：transform.writableNeedDrain的状态是动态变化的，你在判断后立刻执行写入操作，状态可能已经改变，导致逻辑分支判断不准确。
• Promise健壮性不足：仅监听drain事件，未处理流的error或close事件，若流在等待drain时出错或关闭，对应的Promise会永远处于pending状态。
• 逻辑分支冗余：对队列是否为空、是否需要drain的多分支判断，增加了代码复杂度，也容易引发逻辑漏洞。

修正后的实现方案 #

核心优化点是增加并发控制锁，确保同一时间只有一个写入流程在处理队列，同时简化逻辑并增强Promise的健壮性：

import * as stream from 'node:stream';

const transform = new stream.Transform({
    async transform(chunk: Buffer, encoding: BufferEncoding, callback: stream.TransformCallback) {
        callback(null, chunk.toString('utf-8'));
    }
});

stream.pipeline(transform, process.stdout, (err) => {
    if (err) console.error('Pipeline异常:', err);
});

const queue: Array<Buffer> = [];
let isWriting = false; // 并发控制锁，标记是否正在处理写入

async function write(data: Buffer, encoding?: BufferEncoding): Promise<void> {
    // 先将数据加入队列，统一后续处理
    queue.push(data);
    
    // 若当前无写入操作，启动队列处理流程
    if (!isWriting) {
        isWriting = true;
        try {
            while (queue.length > 0 && !transform.closed) {
                const currentChunk = queue[0]; // 先保留队首，写入成功后再移除
                const canWriteMore = transform.write(currentChunk);

                if (!canWriteMore) {
                    // 等待drain，同时监听错误和关闭事件避免Promise挂起
                    await new Promise((resolve, reject) => {
                        const cleanup = () => {
                            transform.off('drain', onDrain);
                            transform.off('error', onError);
                            transform.off('close', onClose);
                        };
                        const onDrain = () => {
                            cleanup();
                            resolve();
                        };
                        const onError = (err: Error) => {
                            cleanup();
                            reject(err);
                        };
                        const onClose = () => {
                            cleanup();
                            resolve();
                        };

                        transform.once('drain', onDrain);
                        transform.once('error', onError);
                        transform.once('close', onClose);
                    });
                }

                // 写入成功，移除队首数据
                queue.shift();
            }
        } catch (err) {
            console.error('写入失败:', err);
            queue.length = 0; // 可根据需求选择清空队列或保留未写入数据
        } finally {
            isWriting = false; // 释放锁，允许后续写入操作启动
        }
    }
}

// 测试调用（无需await也能保证串行处理，await可控制写入节奏）
(async () => {
    for (let i = 0; i < 1e1; i++) {
        await write(Buffer.from('Something'.repeat(1e1), 'utf-8'));
    }
})();

关键优化说明 #

• 并发控制：通过isWriting锁确保同一时间只有一个函数实例处理队列，彻底避免并发竞态问题。
• 简化逻辑：所有写入请求先入队，再统一处理，无需复杂的状态分支判断，代码更易维护。
• 健壮性提升：在等待drain时同时监听流的error和close事件，避免Promise永久挂起，同时捕获异常并处理队列。
• 数据安全：写入前先保留队首数据，确认写入成功后再移除，防止写入失败时数据丢失。

内容的提问来源于stack exchange，提问作者stackhatter