用自己的结果喂芹菜队列

Question

我正在为我的 SPA 应用程序编写爬虫程序。由于它是一个 SPA，我不能使用 wget/curl 或任何其他基于非浏览器的解决方案进行抓取，因为我需要一个浏览器才能在我的 SPA 中运行 javascript。

我使用 python 和 selenium 对此进行了编码。它将从主页开始，扫描所有 href 元素，将它们保存在 set 中，丢弃我已经访问过的那些（visited 和 opened with selenium and collected all the href elements 一样），然后从集合中获取下一个 URL并参观它。然后它将一遍又一遍地重复该过程，直到它访问了所有链接。

代码如下所示：

def main():

    ...

    # Here we will be saving all the links that we can find in the DOM of
    # each visited URL
    collected = set()
    collected.add(crawler.start_url)

    # Here we will be saving all the URLs that we have already visited
    visited = set()

    base_netloc = urlparse(crawler.start_url).netloc

    while len(collected):
        url = collected.pop()

        urls = self.collect_urls(url)
        urls = [x for x in urls if x not in visited and urlparse(x).netloc == base_netloc]

        collected = collected.union(urls)
        visited.add(url)

    crawler.links = list(visited)
    crawler.save()

def collect_urls(self, url):
    browser = Browser()
    browser.fetch(url)

    urls = set()
    elements = browser.get_xpath_elements("//a[@href]")
    for element in elements:
        link = browser.get_element_attribute(element, "href")

        if link != url:
            urls.add(link)

    browser.stop()
    return urls

我希望每次调用collect_urls 时都调用一个 Celery 任务，这样它可以在失败时重试，并且还可以加快整个过程（使用多个工作人员）。问题是collect_urls 是从while 内部调用的，这取决于collected 集合，该集合由collect_urls 的结果填充。

我知道我可以使用 delay() 调用 Celery 任务并使用 get() 等待结果，因此我的代码将如下所示：

    while len(collected):
        url = collected.pop()

        task = self.collect_urls.delay(url)
        urls = task.get(timeout=30)

这会将我对 collect_urls 的调用转换为 Celery 任务，如果出现故障，我可以重试，但我仍然无法使用多个工人，因为我需要等待结果的delay()。

我怎样才能重构我的代码，让我可以为collect_urls 使用多个工作人员？

Answer 1

简短的回答，如果您希望以速度为目的进行分发，您必须将一组已访问的网站设置为跨进程安全的结构。例如，您可以通过将其作为集合存储在 redis 或数据库表中来做到这一点。完成此操作后，您可以更新代码以执行以下操作：

# kick off initial set of tasks:
result_id = uuid.uuid4()
for x in collected:
    task = self.collect_urls.delay(x, result_id)
return result_id

您可以使用该 result_id 定期检查访问的 url 集。一旦该集合在n 的调用次数中具有相同的长度，您就认为它已经完成了。

在 collect_urls 函数中，您实际上是这样做的：

def collect_urls(self, url, result_id):
    # for example, you can use redis smember to check if the 
    # set at result_id contains url
    if url has been visited:
        return
    # you can do this in redis using sadd
    add url to the set of visited
    # collect urls as before
    ...
    # but instead of returning the urls, you kick off new tasks
    for x in urls:
        collect_urls.delay(x, result_id)

如果你使用redis，所有收集/访问的url都将包含在result_id标识的redis键中。您不必使用 redis，您可以轻松地使用数据库中的行来执行此操作，其中 result_id 作为一列，而 url 在另一列中。