如何在循环中通过异步 IO 使用多个选项卡？

Question

我正在尝试使用一定数量的选项卡来打开和操作这些选项卡，同时等待网络 IO 加速我对域的抓取过程。

我已经通过在通道上侦听的工作池解决了 Go 中的相同问题，但我不确定如何在 Node.js 和 Puppeteer 中解决相同的问题。

我的猜测是循环通过

async function nextPage() {
    try {
        for (var link of uncrawledLinks.keys()) {
            if (runningThreads < maxThreads) {
                var page = await browser.newPage();
                console.log("nextPage() # runningThreads: " + runningThreads + " #  uncrawledLinks.size: " + uncrawledLinks.size);
            //debugger;
            crawlPage(page, link);
        }
    }

当我将“maxthreads”增加到高于 1 时可能会产生问题，但直到现在我无法解决问题。

当我将它设置为高于 1 时出现的问题是链接被抓取两次（但不是每个链接，只有大约 90% 的链接），所以我得到重复，这使得爬虫无法使用。

我曾考虑过使用 Redis 或 SQLite 之类的数据库，但我想先解决问题而不使用它以更好地理解问题（直到现在我还没有性能/内存问题，所以在内存中进行操作是不行的问题）。

可运行的示例代码：

'use strict';
const puppeteer = require('puppeteer');
const url = require('url');

// start URL
const startUrlObj = url.parse("http://example.de/");
const startUrlDomain = startUrlObj.protocol + "//" + startUrlObj.hostname;
const startUrl = url.format(startUrlObj);

let browser;
let pages = [];
let uncrawledLinks = new Map();
let crawledLinks = [];
let runningThreads = 0;
const maxThreads = 1;

start();

async function start() {
    console.log("Starting Crawler");
    browser = await puppeteer.launch();
    console.log("Finished initializing browser object");
    uncrawledLinks.set(startUrl, "");
    nextPage();
};

async function crawlPage(page, link) {
    try {
        console.log("starting crawl for: " + link);
    runningThreads++;

    const response = await page.goto(link, {
        waitUntil: 'networkidle2',
        timeout: 30000
    });

    // find all links in the form <a href="xxx">
    const hrefs = await page.$$eval('a', as => as.map(a => a.href));
    hrefs.forEach(function(foundLink, key) {
        if (foundLink.startsWith(startUrlDomain)) {
            var tempUrl = url.parse(foundLink);
                // remove #asd and ?param1=y values from URL
                tempUrl.hash = null;
                tempUrl.search = null;
               var tempLink = url.format(tempUrl);
                //console.log(url.format(tempLink));
            if (crawledLinks.includes(tempLink) === false) {
                if (tempLink.endsWith(".html") === true) {
                    uncrawledLinks.set(tempLink, "false");
                    //pages.push(tempLink);
                }      
        }
    }},  hrefs)
    //console.log("Found new links: " + i + " # " + link);

    // crawling queues
    uncrawledLinks.delete(link);
    crawledLinks.push(link);

} catch (error) {
    // Log errors
    console.error(error);
} finally {
    runningThreads--;
    await page.close();
    await nextPage();
}

}

async function nextPage() {
    try {
        for (var link of uncrawledLinks.keys()) {
            if (runningThreads < maxThreads) {
                var page = await browser.newPage();
                //console.log("nextPage() # runningThreads: " + runningThreads + " #  uncrawledLinks.size: " + uncrawledLinks.size);
            //debugger;
            crawlPage(page, link);
        }
    }


} catch (error) {
    console.error(error);
} finally {
    if (uncrawledLinks.size === 0 && runningThreads === 0) {
        console.log("Finished crawling");
        console.log(crawledLinks);
        await browser.close();
    }
} 
}

Answer 1

首先，不幸的是（或幸运的是，取决于您的观点）您在 Node.js 中没有线程（从技术上讲，您有，但它们在 JavaScript 代码中不可用 - 仅在 C++ 级别上）。这只是为了说明清楚，因为它可能是一个无辜的命名错误或误解，可能会给您带来不合理的期望。

其次，请注意您正在捕获erorr，但打印的是error，因此您可能会遇到一些由于变量拼写错误而看不到的错误。

最后，使用 Bluebird 的 eachLimit 方法可以很容易地解决您正在尝试做的事情（异步操作与在任何给定时间内有多少未完成操作的限制并行）：

• https://caolan.github.io/async/docs.html#eachLimit

使用async 关键字的简单循环并不容易做到这一点，因为通过在循环中使用async，您将停止迭代，直到promise 得到解决，而不是并行执行其他迭代，您必须手动跟踪计数器，但通过awaiting 跟踪计数器的状态而不是给定的函数调用。使用async 模块当然更容易，而不是async 关键字。

Answer 2

我发现必须用 Promise 包装 IO 内容，因为只有 Promise 才能提供异步 IO。

async function crawlPage(page, link) {
    browser.newPage().then(
        async page => {
            try {
                await page.setRequestInterception(true);
                // catch all requests
                page.on('request', request => {

                })

                await page.goto(uncrawledUrl, {
                    waitUntil: 'networkidle2',
                    timeout: 10000
                }).then(
                    async response => {