Node.js异步场景下文件导入计数器值异常问题求助

问题代码 #

app.post("/upload", upload.array("files"), uploadFiles);
function uploadFiles(req, res) {
    req.files.forEach(function(file) {
        var linesInserted = 0;
        var linesInError = 0;
        fs.createReadStream(file.destination + file.filename)
        .pipe(parse({ delimiter: ",", columns: false, fromLine: 2 }))
        .on("data", function (row) {
            Model.findOneAndUpdate(
                { code: row[0] },
                { 
                    $set: { 
                        a: row[1],
                        b: row[2],
                        c: row[4],
                        d: row[5],
                        e: row[6],
                        f: moment(row[7], "DD/MM/YYYY").toISOString(),
                        g: moment(row[8], "DD/MM/YYYY").toISOString(),
                        h: row[12].split(/\s/)[0].replace(',','.')
                    },
                    $setOnInsert: { 
                        i: row[0], 
                        j: "airbnb",
                        k: row[3],
                        l: moment(row[10], "DD/MM/YYYY").toISOString()
                    },
                    $push: { 
                        connectedTo: [{ m : "xxx" }, { service: "n", serviceId: "o" }] 
                    }
                },
                { 
                    upsert: true,
                    runValidators: true
                },
                function(err, res) {
                    if (err) {
                        console.log(err);
                        linesInError++;
                    } else if (res) {
                        console.log(linesInserted);
                        linesInserted++;
                    }
                }
            );
        })
        .on("end", function () {
            File.create({
                file: file.destination + file.filename,
                originalName: file.originalname,
                linesInserted: linesInserted,
                linesInError: linesInError
            });

            console.log(`File ${file.originalname} - ${file.destination + file.filename} imported`);
        })
        .on("error", function (error) {
            console.log(error.message);
        });
    });
    res.json({ counterFilesImported: req.files.length });
}

问题描述 #

读取CSV文件并通过findOneAndUpdate写入数据库时，end事件触发后创建File文档，linesInserted和linesInError始终为0。虽然读取过程中计数器会递增，但end事件执行时这些异步的数据库操作还没完成，导致计数器还是初始值。

解决方案 #

核心思路是等待所有异步数据库操作完成后，再统计数量并创建File文档，具体步骤如下：

1. 将findOneAndUpdate转为Promise形式（Mongoose方法不传回调会返回Promise）
2. 收集每一行的数据库操作Promise到数组中
3. 在end事件中，用Promise.allSettled等待所有操作完成，统计成功和失败的数量
4. 最后再创建File文档

修改后的代码 #

app.post("/upload", upload.array("files"), uploadFiles);
async function uploadFiles(req, res) {
    // 遍历每个上传的文件
    for (const file of req.files) {
        const operations = [];
        fs.createReadStream(file.destination + file.filename)
            .pipe(parse({ delimiter: ",", columns: false, fromLine: 2 }))
            .on("data", (row) => {
                // 将findOneAndUpdate转为Promise，加入操作数组
                const op = Model.findOneAndUpdate(
                    { code: row[0] },
                    {
                        $set: {
                            a: row[1],
                            b: row[2],
                            c: row[4],
                            d: row[5],
                            e: row[6],
                            f: moment(row[7], "DD/MM/YYYY").toISOString(),
                            g: moment(row[8], "DD/MM/YYYY").toISOString(),
                            h: row[12].split(/\s/)[0].replace(',', '.')
                        },
                        $setOnInsert: {
                            i: row[0],
                            j: "airbnb",
                            k: row[3],
                            l: moment(row[10], "DD/MM/YYYY").toISOString()
                        },
                        $push: {
                            connectedTo: [{ m: "xxx" }, { service: "n", serviceId: "o" }]
                        }
                    },
                    {
                        upsert: true,
                        runValidators: true
                    }
                )
                .then(() => true) // 成功标记为true
                .catch((err) => {
                    console.log(err);
                    return false; // 失败标记为false
                });
                operations.push(op);
            })
            .on("end", async () => {
                // 等待所有操作完成
                const results = await Promise.allSettled(operations);
                // 统计成功和失败数量
                const linesInserted = results.filter(r => r.status === 'fulfilled' && r.value).length;
                const linesInError = results.filter(r => r.status === 'rejected' || !r.value).length;
                
                // 创建File文档
                await File.create({
                    file: file.destination + file.filename,
                    originalName: file.originalname,
                    linesInserted,
                    linesInError
                });

                console.log(`File ${file.originalname} - ${file.destination + file.filename} imported`);
            })
            .on("error", (error) => {
                console.log(error.message);
            });
    }
    res.json({ counterFilesImported: req.files.length });
}

关键说明 #

• 使用Promise.allSettled而不是Promise.all：前者会等待所有操作完成（不管成功失败），不会因为单个操作失败而中断，适合统计全量结果
• 把uploadFiles改为async函数，在end事件的回调中使用await等待Promise结果
• 每个数据库操作返回标记值，用于后续统计成功/失败数量

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