在 Celery 任务中运行 Scrapy 蜘蛛

Question

This is not working anymore，scrapy 的 API 变了。

现在文档提供了“Run Scrapy from a script”的方法，但我收到了ReactorNotRestartable 错误。

我的任务：

from celery import Task

from twisted.internet import reactor

from scrapy.crawler import Crawler
from scrapy import log, signals
from scrapy.utils.project import get_project_settings

from .spiders import MySpider



class MyTask(Task):
    def run(self, *args, **kwargs):
        spider = MySpider
        settings = get_project_settings()
        crawler = Crawler(settings)
        crawler.signals.connect(reactor.stop, signal=signals.spider_closed)
        crawler.configure()
        crawler.crawl(spider)
        crawler.start()

        log.start()
        reactor.run()

Answer 1

扭曲的反应堆无法重新启动。解决此问题的方法是让 celery 任务为您要执行的每个爬网创建一个新的子进程，如以下帖子中所建议的那样：

• Running Scrapy spiders in a Celery task

这通过使用multiprocessing 包解决了“reactor 无法重新启动”问题。但问题在于，由于您将遇到另一个问题，即守护进程无法生成子进程，因此最新的 celery 版本现在已经过时了。因此，为了使解决方法起作用，您需要使用 celery 版本。

是的，scrapy API 已更改。但稍作修改（import Crawler 而不是CrawlerProcess）。您可以通过使用 celery 版本来获得解决方法。

可以在此处找到 Celery 问题： Celery Issue #1709

这是我的更新的爬虫脚本，通过使用billiard 而不是multiprocessing，与较新的芹菜版本一起工作：

from scrapy.crawler import Crawler
from scrapy.conf import settings
from myspider import MySpider
from scrapy import log, project
from twisted.internet import reactor
from billiard import Process
from scrapy.utils.project import get_project_settings
from scrapy import signals


class UrlCrawlerScript(Process):
    def __init__(self, spider):
        Process.__init__(self)
        settings = get_project_settings()
        self.crawler = Crawler(settings)
        self.crawler.configure()
        self.crawler.signals.connect(reactor.stop, signal=signals.spider_closed)
        self.spider = spider

    def run(self):
        self.crawler.crawl(self.spider)
        self.crawler.start()
        reactor.run()

def run_spider(url):
    spider = MySpider(url)
    crawler = UrlCrawlerScript(spider)
    crawler.start()
    crawler.join()

编辑：通过阅读 celery 问题#1709，他们建议使用台球而不是多处理，以便解除子进程限制。换句话说，我们应该尝试billiard 看看它是否有效！

编辑 2： 是的，通过使用 billiard，我的脚本适用于最新的 celery 版本！查看我更新的脚本。

Answer 2

Twisted reactor 无法重新启动，因此一旦一个蜘蛛程序完成运行并且crawler 隐式停止了 reactor，该 worker 将毫无用处。

正如在另一个问题的答案中所发布的那样，您需要做的就是杀死运行您的蜘蛛的工人并用一个新的替换它，这样可以防止反应堆多次启动和停止。为此，只需设置：

CELERYD_MAX_TASKS_PER_CHILD = 1

缺点是您并没有真正使用 Twisted reactor 发挥其全部潜力并浪费运行多个反应器的资源，因为一个反应器可以在一个进程中同时运行多个蜘蛛。更好的方法是每个工人运行一个反应器（甚至全球一个反应器）并且不要让crawler 触摸它。

我正在为一个非常相似的项目做这个，所以如果我有任何进展，我会更新这篇文章。

Answer 3

为了避免在 Celery 任务队列中运行 Scrapy 时出现 ReactorNotRestartable 错误，我使用了线程。在一个应用程序中多次运行 Twisted reactor 的方法相同。 Scrapy 也使用了 Twisted，所以我们可以这样做。

代码如下：

from threading import Thread
from scrapy.crawler import CrawlerProcess
import scrapy

class MySpider(scrapy.Spider):
    name = 'my_spider'


class MyCrawler:

    spider_settings = {}

    def run_crawler(self):

        process = CrawlerProcess(self.spider_settings)
        process.crawl(MySpider)
        Thread(target=process.start).start()

不要忘记为 celery 增加 CELERYD_CONCURRENCY。

CELERYD_CONCURRENCY = 10

对我来说很好。

这不是阻塞进程运行，但无论如何，scrapy 的最佳实践是在回调中处理数据。这样做：

for crawler in process.crawlers:
    crawler.spider.save_result_callback = some_callback
    crawler.spider.save_result_callback_params = some_callback_params

Thread(target=process.start).start()

Answer 4

如果您有很多任务要处理，我会说这种方法非常低效。 因为 Celery 是线程化的 - 在它自己的线程中运行每个任务。 假设使用 RabbitMQ 作为代理，您可以通过 >10K q/s。 使用 Celery，这可能会导致 10K 线程开销！ 我建议不要在这里使用芹菜。而是直接访问代理！