这是为我做的工作-你能描述一下如何用你的框架实现它吗-我相信应该有办法，我只需要正确地配置它

//use fast-csv to stream data from a file
    csv
      .fromPath(form.FileName, { headers: true })
      .on("data", async data => {
        const query = await queryBuilder({
          schema,
          routine,
          parameters,
          request
        }); //here we prepare query for calling the SP with parameters from data
        winston.info(query + JSON.stringify(data));
        const rawResponse = await session.raw(query); //here the query gets executed
        fileRows.push(data); // push each row - for testing only
      })
      .on("end", function() {
        console.log(fileRows);
        fs.unlinkSync(form.FileName); // remove temp file
        //process "fileRows" and respond
        res.end(JSON.stringify(fileRows)) // - for testing
      });

正如在评论中提到的，我让我的超燃冲压发动机轻松处理这样一个用例。。。请纠正我，如果我理解错了，但我知道你想调用两个等待行为每个csv行在测试中。
如果是这样，您的代码将如下所示（更新以匹配您的注解/答案）：

var fs = require("fs");
var csv = require("fast-csv");
var stream1 = fs.createReadStream("files/testCsvFile.csv");
var {DataStream} = require("scramjet");
DataStream
    // the following line will convert any stream to scramjet.DataStream
    .from(csv.fromStream(stream2, { headers: true }))
    // the next lines controls how many simultaneous operations are made
    // I assumed 16, but if you're fine with 40 or you want 1 - go for it.
    .setOptions({maxParallel: 16})
    // the next line will call your async function and wait until it's completed
    // and control the back-pressure of the stream
    .do(async (data) => {
        const query = await queryBuilder({
          schema,
          routine,
          parameters,
          request
        }); //here we prepare query for calling the SP with parameters from data
        winston.info(query + JSON.stringify(data));
        const rawResponse = await session.raw(query); //here the query gets executed
        return data; // push each row - for testing only)
    })
    // next line will run the stream until end and return a promise
    .toArray()
    .then(fileRows => {
        console.log(fileRows);
        fs.unlinkSync(form.FileName); // remove temp file
        //process "fileRows" and respond
        res.end(JSON.stringify(fileRows)); // - for testing
    })
    .catch(e => {
        res.writeHead(500); // some error handling
        res.end(e.message);
    })
    ;
    // you may want to put an await statement before this, or call then to check
    // for errors, which I assume is your use case.
;

回答您的评论问题-如果您在 on("data") 事件-您需要创建一个promises数组并等待promise。所有这些数组都在流中 end -但这需要同步完成，所以事件处理程序中的异步函数不能做到这一点。
在 scramjet 这发生在引擎盖下，所以你可以使用这个功能。