在Celery任务中运行Scrapy蜘蛛

Question

我有一个Django站点,当用户请求它时会发生刮擦,我的代码在新进程中启动Scrapy蜘蛛独立脚本.当然,这不会增加用户.

像这样的东西:

class StandAloneSpider(Spider):
    #a regular spider

settings.overrides['LOG_ENABLED'] = True
#more settings can be changed...

crawler = CrawlerProcess( settings )
crawler.install()
crawler.configure()

spider = StandAloneSpider()

crawler.crawl( spider )
crawler.start()

我决定使用Celery并使用worker来排队爬网请求.

但是,我遇到了Tornado反应堆无法重启的问题.第一个和第二个蜘蛛成功运行,但后续的蜘蛛会抛出ReactorNotRestartable错误.

任何人都可以在Celery框架内分享运行蜘蛛的任何提示？

Answer 1

好的,这就是我如何让Scrapy使用我的Django项目,该项目使用Celery来排队爬行的内容.实际的解决方法主要来自joehillen的代码http://snippets.scrapy.org/snippets/13/

首先是tasks.py文件

from celery import task

@task()
def crawl_domain(domain_pk):
    from crawl import domain_crawl
    return domain_crawl(domain_pk)

然后是crawl.py文件

from multiprocessing import Process
from scrapy.crawler import CrawlerProcess
from scrapy.conf import settings
from spider import DomainSpider
from models import Domain

class DomainCrawlerScript():

    def __init__(self):
        self.crawler = CrawlerProcess(settings)
        self.crawler.install()
        self.crawler.configure()

    def _crawl(self, domain_pk):
        domain = Domain.objects.get(
            pk = domain_pk,
        )
        urls = []
        for page in domain.pages.all():
            urls.append(page.url())
        self.crawler.crawl(DomainSpider(urls))
        self.crawler.start()
        self.crawler.stop()

    def crawl(self, domain_pk):
        p = Process(target=self._crawl, args=[domain_pk])
        p.start()
        p.join()

crawler = DomainCrawlerScript()

def domain_crawl(domain_pk):
    crawler.crawl(domain_pk)

这里的技巧是"来自多处理导入过程",它解决了Twisted框架中的"ReactorNotRestartable"问题.因此,基本上Celery任务调用"domain_crawl"函数,该函数反复使用"DomainCrawlerScript"对象来与Scrapy蜘蛛进行交互.(我知道我的例子有点多余,但我确实这样做是因为我的设置有多个版本的python [我的django webserver实际上是使用python2.4而我的工作服务器使用python2.7])

在我的例子中,"DomainSpider"只是一个修改过的Scrapy Spider,它将一个url列表设置为"start_urls".

希望这可以帮助!

Answer 2

我在设置文件中将CELERYD_MAX_TASKS_PER_CHILD设置为1并处理了该问题.每个蜘蛛运行后,工作器守护程序启动一个新进程并负责处理反应器.