Node.js管道向MongoDB发送空对象问题求助

核心问题分析 #

1. async与callback混用冲突：Transform流的transform方法同时使用async修饰和手动调用callback会导致异步逻辑混乱——async函数会返回Promise，而流会同时等待Promise完成和callback调用，这会触发错误的状态判断，最终导致空对象写入或流处理异常。
2. pipeline调用方式错误：你给pipeline加了async修饰符，但pipeline本身就返回Promise，正确的做法是用await直接调用它，而非标记pipeline为异步函数。
3. csvtojson参数传递错误：csvtojson的选项不需要拆成两个对象传递，正确的参数格式是直接传入配置对象。

修复步骤 #

1. 修正Transform的异步处理逻辑 #

选择以下两种方式之一：

方式一：使用Promise链式调用替代async/await + callback

const writeToMongoDB = new Transform({
    objectMode: true,
    transform(chunk: IData, enc, callback) {
        console.log(chunk)
        ColorModel.create(chunk)
            .then(() => callback(null)) // 成功时调用callback，无返回数据
            .catch(err => callback(err)) // 失败时传递错误
    },
})

方式二：纯async/await（无需手动调用callback）

从Node.js 10+开始，Transform流支持async类型的transform方法，此时无需调用callback，流会自动等待Promise完成，错误也会被自动捕获传递：

const writeToMongoDB = new Transform({
    objectMode: true,
    async transform(chunk: IData, enc) {
        console.log(chunk)
        await ColorModel.create(chunk)
        // 如果需要向下游传递数据，直接return chunk即可；不需要则省略
    },
})

2. 修复pipeline调用 #

移除pipeline前的async，改用await直接调用：

await pipeline(
    readableStream,
    csvtojson({ delimiter: ',' }), // 修正参数格式
    writeToMongoDB,
    (err) => {
        if (err) console.error('管道处理失败:', err)
    }
)

3. 可选：移除无用代码 #

你的代码中myTransformToJSON未被使用，可以直接删除。

完整修复后代码 #

import { Transform, pipeline } from 'stream'
import fs from 'fs'
import csvtojson from 'csvtojson'
import { connect } from './your-mongo-connect-module' // 替换为实际导入路径
import ColorModel from './your-color-model' // 替换为实际导入路径

async function main() {
    await connect(MONGODB_ACCESS_KEY as string)
    const readableStream = fs.createReadStream('./color_srgb.csv')

    const writeToMongoDB = new Transform({
        objectMode: true,
        async transform(chunk: IData, enc) {
            console.log(chunk)
            await ColorModel.create(chunk)
        },
    })

    try {
        await pipeline(
            readableStream,
            csvtojson({ delimiter: ',' }),
            writeToMongoDB
        )
        console.log('数据写入完成')
    } catch (err) {
        console.error('管道执行出错:', err)
    }
}

main().catch(err => console.error('主函数执行失败:', err))

注：这里把pipeline的回调换成了try/catch，用Promise的方式处理错误更符合async/await的风格，你也可以保留回调写法，二选一即可。

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