如何从第三方API逐次返回的JSON对象创建可读流实现CSV导出

• 流结束信号时序错误：rs.push(null) 是同步执行代码，直接写在异步游标遍历的外部，程序运行时会先初始化流、推送结束标记、搭好管道链路，这时候游标还没开始从数据库拉取数据，流已经被判定为完成状态，后续循环中推入的所有业务数据根本无法通过管道传输
• 自实现可读流无背压适配：rs._read = () => {} 本质是空壳流，不管下游客户端的下载速度多慢，会无限制把待导出的数据堆在内存里，数据量稍大就会触发内存溢出
• 管道链路选型完全错位：目标是生成CSV文件，链路中却接入了JSONStream.stringify()、jsonParser两类仅用于JSON格式处理的组件，就算流通了，输出内容也是错乱的JSON片段，无法生成可用的CSV文件
• 全链路无错误捕获：populate请求失败、游标查询异常、客户端提前断连等场景都没有做监听，一旦触发错误不是请求永久挂死就是进程崩溃

实现思路很直接：弃用空壳的自实现Readable流，直接基于Mongoose游标逐行遍历处理数据，对接专门的CSV转换流，对齐执行时序、做好背压控制和错误兜底。

const { format } = require('@fast-csv/format');
const fileName = `Data.csv`;

// 原有HTTP响应头配置保留
res.set('Content-Type', 'text/csv');
res.setHeader('Content-disposition', 'attachment; filename=' + fileName);
res.flushHeaders();

// 初始化CSV转换流，自动处理列头、格式转义
const csvStream = format({
    headers: true,
    writeHeaders: true
});

// 初始化Mongoose查询游标
const itemCursor = OfferItem.find(query)
    .read('sp')
    .sort('purchase_date')
    .cursor();

// 逐行处理游标返回的数据
itemCursor.eachAsync(async (offerItemData) => {
    // 暂停游标拉取，当前行处理完成后再恢复，控制内存占用
    itemCursor.pause();

    try {
        // 原有populate逻辑保留
        offerItemData = await UserService.populateUser(offerItemData, {
            path: 'purchased_by',
            select: '_id first_name last_name email phone',
            options: {
                readPreference: 'secondaryPreferred'
            }
        });

        offerItemData = await UserAddressService.populateUserAddress(offerItemData, {
            path: 'delivery.user_address',
            options: {
                readPreference: 'secondaryPreferred'
            }
        });

        // 原有字段转换逻辑，补充空值兜底避免脏数据报错
        const sku = offerItemData.sku_id ? offerItemData.sku_id : offerItemData._id;
        const user = offerItemData.purchased_by || {};
        const name = humanize.capitalizeAll(`${user.first_name || ''} ${user.last_name || ''}`);
        const { email, phone } = user;
        const coin = offerItemData.coin_price;
        let address = '';
        let pinCode = '';
        let code = '';

        const deliveryAddress = offerItemData.delivery?.user_address;
        if (deliveryAddress) {
            address = `${deliveryAddress.basic || ''}, ${deliveryAddress.area || ''}, ${deliveryAddress.city_str || ''}`;
            pinCode = deliveryAddress.pin_code;
        }

        if (offerItemData.discount) {
            code = offerItemData.discount.code;
        }

        const data = {
            sku,
            name,
            address,
            pinCode,
            phone,
            email,
            code,
            coin
        };

        // 把转换好的对象写入CSV流
        csvStream.write(data);

    } catch (err) {
        // 单条数据处理出错仅记录日志，不中断整个导出流程
        console.error('单条订单数据处理失败', err);
    }

    // 当前行处理完成，恢复游标拉取下一条数据
    itemCursor.resume();
}).then(() => {
    // 游标完全遍历结束后，结束CSV流，触发管道传输完成
    csvStream.end();
}).catch((err) => {
    // 游标遍历整体出错，销毁流并传递错误
    csvStream.destroy(err);
});

// 组装管道链路：CSV流直接写入HTTP响应
csvStream.pipe(res);

// 全链路错误监听
csvStream.on('error', (err) => {
    res.destroy(err);
});

// 客户端提前断开连接时，关闭数据库游标，避免无效拉取
res.on('close', () => {
    itemCursor.close();
});

关键调整说明 #

• 替换原有的JSON处理组件，使用专门的CSV格式流处理库，自动处理CSV列头、特殊字符转义、换行分隔等格式问题，无需手动拼接CSV字符串
• 通过游标pause/resume控制数据拉取节奏，和流的背压机制对齐，避免全量数据堆在内存中导致的OOM问题
• 把流结束逻辑移到游标遍历完成的Promise回调中，等所有数据处理写入完成后再推送结束标记，解决原来流提前终止的问题
• 补充空值兜底逻辑，避免单条数据关联的用户、收货地址为空时抛出异常，中断整个导出流程
• 增加响应关闭事件监听，客户端取消下载时直接终止数据库查询，减少不必要的资源消耗

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