等待文件加载，然后在每一行上运行一个函数，但等待每一行的返回

Question

我一直在尝试理解 Promise，但我遇到了障碍。

==Order 我要代码运行==

1. 我需要一个 .txt 文件来将每一行加载到一个数组中。
2. 等待这一切发生。
3. 对每个返回数组的条目运行一个函数。
4. 等待要处理的数组的每个索引，然后再执行下一个。

==我的功能==

1. 调用该函数启动程序。

async function start(){
    var data = await getData();
    console.log(data);
    for (var i = 0; i < data.length; i++){
        console.log(await searchGoogle(data[i]));
    }

}

1. '等待'来自 getData 的数据

async function getData(){
    return new Promise(function(resolve, reject){
        fs.readFile('./thingsToGoogle.txt', function(err, data) {
            if(err) throw err;
            var array = data.toString().split("\n");
            resolve(array); 
        });
    });
}

1. 然后对数组中的每个索引调用 searchGoogle。

async function searchGoogle(toSearch) {
    (async() => {
        const browser = await puppeteer.launch();
        const page = await browser.newPage();
        await page.goto('https://www.google.com/');
        await page.type('input[name=q]', toSearch);
        try {
            console.log('Setting Search' + toSearch);           
            await page.evaluate(() => {
                let elements = document.getElementsByClassName('gNO89b');
                for (let element of elements)
                    element.click();
            });
            await page.waitForNavigation();         
        } catch (err) {
            console.log(err)
        }

        try {
            console.log("Collecting Data");
            const[response] = await Promise.all([
                        page.waitForNavigation(),
                        await page.click('.rINcab'),
                    ]);
        } catch (err) {
            console.log("Error2: " + err)
        }

        let test = await page.$$('.LC20lb');
        // console.log(test);
        allresults = [];
        for (const t of test) {
            const label = await page.evaluate(el => el.innerText, t);
            if (label != "") {
                allresults.push(label);
            }
        }
        await browser.close();
        resolve(allresults);

    })();
}

问题是这不起作用。它不会等待文件加载。

Picture of Node JS output.

希望屏幕截图已上传，但您可以看到它堆叠了 SearchGoogle 函数 console.logs；

console.log('Setting..')
console.log('Setting..')
console.log('Collecting..')
console.log('Collecting..')

应该是什么时候

console.log('Setting..')
console.log('Collecting..')
console.log('Setting..')
console.log('Collecting..')

这是第一次处理 Promise，我对它们进行了大量阅读并编写了一些代码来理解它们，但是当我尝试应用这些知识时，我感到很挣扎。希望有人能帮忙。

-桃人-

Answer 1

具有并发限制的队列（使用p-queue）

您需要一个具有并发限制的队列。您将阅读每一行并将它们添加到队列中。我们将为此使用readline 和p-queue 模块。

首先，创建一个并发为1的队列。

const {default: PQueue} = require('p-queue');
const queue = new PQueue({concurrency: 1});

然后，创建我们的阅读器实例。

const fs = require('fs');
const readline = require('readline');

const rl = readline.createInterface({
  input: fs.createReadStream('your-input-file.txt')
});

对于文件的每一行，将一个条目添加到队列中。

rl.on('line', (line) => {
  console.log(`Line from file: ${line}`);
  queue.add(() => searchGoogle(line));
});

就是这样！如果要一次处理 10 行，只需更改并发行。它仍然会一次读取一行，但队列会限制调用多少个searchGoogle。

可选修复：异步等待

您的代码具有以下结构，

async yourFunction(){
  (async()=>{
    const browser = await puppeteer.launch();
    // ... rest of the code
  })()
}

虽然这可能会按预期运行，但您将很难调试，因为每次运行 yourFunction 时都会创建一个匿名函数。

以下就足够了。

async yourFunction(){
    const browser = await puppeteer.launch();
    // ... rest of the code
}

Answer 2

这是一种处理它们的方法，可让您一次处理 N 个 URL，您可以调整 N 的值。我的猜测是，您希望将其设置为 5 到 20 之间的值，以保持 CPU 繁忙，但不要使用过多的服务器资源。

以下是其工作原理的概要：

1. 它使用 line-by-line 模块逐行读取文件（与内置的 readline 接口不同），当您调用 .pause() 时，此模块会暂停 line 事件，这在此实现中很重要。
2. 它维护一个numInFlight 计数器，告诉您有多少行正在处理中。
3. 您将maxInFlight 常量设置为您希望并行处理的最大行数。
4. 它维护一个resultCntr，可帮助您保持结果的正确顺序。
5. 它创建 readline 接口并为line 事件建立一个监听器。这将启动带有line 事件的流。
6. 在每个line 事件中，我们都会增加numInFlight 计数器。如果我们达到了飞行中允许的最大数量，我们会暂停 readline 流，这样它就不会再产生任何line 事件。如果我们在飞行中还没有达到最大值，那么更多的line 事件将会继续，直到我们达到最大值。
7. 我们将该行传递给您现有的 searchGoogle() 函数。
8. 处理完该行后，我们将结果保存在数组中的适当位置，递减 numInFlight 计数器并恢复流（以防它之前暂停）。
9. 我们检查是否全部完成（通过检查numInFlight 是否为0 以及是否已到达文件末尾）。如果我们完成了，请使用结果解决主 Promise。
10. 如果我们还没有完成，那么将会有更多 line 事件即将到来，或者更多 searchGoogle() 正在运行的函数将完成，这两个函数都将再次检查我们是否已完成。李>
11. 请注意，此设计的工作方式是将任何给定 URL 上的错误放入结果数组中（错误对象在数组中），并继续对其余 URL 进行处理，最终得到解决的承诺。读取输入文件时出错将终止处理并拒绝返回承诺。

代码如下：

const fs = require('fs');
const Readline = require('line-by-line');

function searchAll(file) {
    return new Promise(function(resolve, reject) {
        const rl = new Readline(file);
        // set maxInFlight to something between 5 and 20 to optimize performance by
        // running multiple requests in flight at the same time without
        // overusing memory and other system resources.
        const maxInFlight = 1;

        let numInFlight = 0;
        let resultCntr = 0;
        let results = [];
        let doneReading = false;

        function checkDone(e) {
            if (e) {
                reject(e);
            } else if (doneReading && numInFlight === 0) {
                resolve(results);
            }
        }

        rl.on('line', async (url) => {
            if (url) {
                let resultIndex = resultCntr++;
                try {
                    ++numInFlight;
                    if (numInFlight >= maxInFlight) {
                        // stop flowing line events when we hit maxInFlight
                        rl.pause();
                    }
                    let result = await searchGoogle(url);
                    // store results in order
                    results[resultIndex] = result;
                } catch(e) {
                    // store error object as result
                    results[resultIndex] = e;
                } finally {
                    --numInFlight;
                    rl.resume();
                    checkDone();
                }
            }
        }).on('end', () => {
            // all done reading here, may still be some processing in flight
            doneReading = true;
            checkDone();
        }).on('error', (e) => {
            doneReading = true;
            checkDone(e);
        });    
    });
}

仅供参考，您可以将 maxInFlight 设置为 1 的值，它会一次读取处理一个 URL，但编写此类函数的全部意义在于，您可以通过以下方式获得更好的性能将其设置为高于 1 的值（我猜是 5-20）。