Async/Await in for loop Node JS 不阻塞循环执行

Question

我知道老派 for 循环以传统方式工作 - 它等待 await 完成获取结果。但在我的用例中，我需要从 local/s3 读取一个文件并逐行处理它，并且对于每一行我都需要调用外部 API。通常我在循环中使用 await 因为所有的都是运行在 lambda 中并且我不想将所有内存用于运行它并行。在这里，我使用 stream.on() 方法读取文件，为了在其中使用 await，我需要在读取方法中添加异步，如下所示：

stream.on('data',async () =>{
         while(data=stream.read()!==null){
           console.log('line');
           const requests = getRequests(); // sync code,no pblms
           for(let i=0;i<requests.length;i++){
             const result = await apiCall(request[i);
             console.log('result from api')
             const finalResult = await anotherapiCall(result.data);
             }
        }
});

这是可行的，但不能保证行的处理顺序。我需要所有同步方式。有帮助吗？

完整代码

async function processSOIFileLocal (options, params) {
console.log('Process SOI file');

const readStream = byline.createStream(fs.createReadStream(key));
readStream.setEncoding('utf8');
const pattern = /^UHL\s|^UTL\s/;
const regExp = new RegExp(pattern);
readStream.on('readable', () => {
    let line;
    while (null !== (line = readStream.read())) {
        if (!regExp.test(line.toString())) {
            totalRecordsCount++;
            dataObject = soiParser(line);
            const { id } = dataObject;
            const XMLRequests = createLoSTRequestXML(
                options,
                { mapping: event.mapping, row: dataObject }
            );
            console.log('Read line');
            console.log(id);
            try {
                for (let i = 0;i < XMLRequests.length;i++) {
                    totalRequestsCount++;
                    console.log('Sending request');
                    const response = await sendLoSTRequest(
                        options,
                        { data: XMLRequests[i],
                            url: LOST_URL }
                    );
                    console.log("got response");
                    const responseObj = await xml2js.
                        parseStringPromise(response.data);
                    if (Object.keys(responseObj).indexOf('errors') !== -1) {
               
                        fs.writeFileSync(`${ERR_DIR}/${generateKey()}-${id}.xml`, response.data);
                        failedRequestsCount++;
                    } else {
                        successRequestsCount++;
                        console.log('Response from the Lost Server');
                        console.log(response[i].data);
                    }
                }
            } catch (err) {
                console.log(err);
            }
        }
    }
})
    .on('end', () => {
        console.log('file processed');
        console.log(`
        ************************************************
        Total Records Processed:${totalRecordsCount}
        Total Requests Sent: ${totalRequestsCount}
        Success Requests: ${successRequestsCount}
        Failed Requests: ${failedRequestsCount}
        ************************************************
        `);
    });
}

async function sendLoSTRequest (options, params) {
const { axios } = options;
const { url, data } = params;
if (url) {
    return  axios.post(url, data);
// eslint-disable-next-line no-else-return
} else {
    console.log('URL is not found');
    return null;
}

}

代码需要像这样流动：以同步方式读取一行处理该行并将该行转换为两个成员的数组每个成员调用 API 并做一些事情一行完成后，寻找另一行，全部按顺序完成

更新：有一个解决方法..但它触发 stream.end() 而不等待流完成读取

async function processSOIFileLocal (options, params) {
console.log('Process SOI file');
const { ERR_DIR, fs, xml2js, LOST_URL, byline, event } = options;
const { key } = params;
const responseObject = {};
let totalRecordsCount = 0;
let totalRequestsCount = 0;
let failedRequestsCount = 0;
let successRequestsCount = 0;
let dataObject = {};
const queue = (() => {
    let q = Promise.resolve();
    return fn => (q = q.then(fn));
})();
const readStream = byline.createStream(fs.createReadStream(key));
readStream.setEncoding('utf8');
const pattern = /^UHL\s|^UTL\s/;
const regExp = new RegExp(pattern);
readStream.on('readable', () => {
    let line;
    while (null !== (line = readStream.read())) {
        if (!regExp.test(line.toString())) {
            totalRecordsCount++;
            dataObject = soiParser(line);
            const { id } = dataObject;
            const XMLRequests = createLoSTRequestXML(
                options,
                { mapping: event.mapping, row: dataObject }
            );
                // eslint-disable-next-line no-loop-func
            queue(async () => {
                try {
                    for (let i = 0;i < XMLRequests.length;i++) {
                        console.log('Sending request');
                        console.log(id);
                        totalRequestsCount++;
                        const response = await sendLoSTRequest(
                            options,
                            { data: XMLRequests[i],
                                url: LOST_URL }
                        );
                        console.log('got response');
                        const responseObj = await xml2js.
                            parseStringPromise(response.data);
                        if (Object.keys(responseObj).indexOf('errors') !== -1) {
                            // console.log('Response have the error:');
                            // await handleError(options, { err: responseObj, id });
                            failedRequestsCount++;
                            fs.writeFileSync(`${ERR_DIR}/${generateKey()}-${id}.xml`, response.data);
                        } else {
                            console.log('Response from the Lost Server');
                            console.log(response[i].data);
                            successRequestsCount++;
                        }
                    }
                } catch (err) {
                    console.log(err);
                }
            });
        }
    }
})
    .on('end', () => {
        console.log('file processed');
        console.log(`
            ************************************************
            Total Records Processed:${totalRecordsCount}
            Total Requests Sent: ${totalRequestsCount}
            Success Requests: ${successRequestsCount}
            Failed Requests: ${failedRequestsCount}
            ************************************************
            `);
        Object.assign(responseObject, {
            failedRequestsCount,
            successRequestsCount,
            totalRecordsCount,
            totalRequestsCount
        });
    });

}

谢谢

Answer 1

问题顶部的示例代码可以重写为

const queue = (() => {
    let q = Promise.resolve();
    return (fn) => (q = q.then(fn));
})();
stream.on('data', async() => {
    while (data = stream.read() !== null) {
        console.log('line');
        const requests = getRequests(); // sync code,no pblms
        queue(async () => {
            for (let i = 0; i < requests.length; i++) {
                const result = await apiCall(request[i]);
                console.log('result from api');
                const finalResult = await anotherapiCall(result.data);
            }
        });
    }
});

希望这对完整代码有用

Answer 2

如果有人想要同步处理文件的解决方案，即逐行读取并执行一些异步调用，建议使用内置流转换。在那里我们可以创建一个转换函数和 return 完成时的回调。这将帮助任何人面对这个问题。 Through2 是一个小型的 npm 库，也可以用于相同的目的。

Async/Await in for loop Node JS 不阻塞循环执行

Async/Await in for loop NodeJS Not blocking the loop execuation

javascript

node.js

async-await

es6-promise

完整代码