Scrapy批量处理2k个URL遇ReactorNotRestartable错误求解决方案

问题背景 #

数据库存储约2000个待爬取URL，原代码计划每次处理10个，但仅第一次循环正常执行，第二次触发twisted.internet.error.ReactorNotRestartable错误。

原代码：

from scrapy.crawler import CrawlerProcess
from scrapy.utils.project import get_project_settings

URLs = crawler_table.find(crawl_timestamp=None)
settings = get_project_settings()
for i in range(len(URLs) // 10):
    process = CrawlerProcess(settings)

    limit = 10
    kount = 0

    for crawl in crawler_table.find(crawl_timestamp=None):
        if kount < limit:
            kount += 1
            process.crawl(
                MySpider,
                start_urls=[crawl['crawl_url']]
           )
    process = CrawlerProcess(settings)
    process.start()

错误日志：

File "C:\Program Files\Python310\lib\site-packages\scrapy\crawler.py", line 327, in start
    reactor.run(installSignalHandlers=False)  # blocking call
File "C:\Program Files\Python310\lib\site-packages\twisted\internet\base.py", line 1314, in run
    self.startRunning(installSignalHandlers=installSignalHandlers)
File "C:\Program Files\Python310\lib\site-packages\twisted\internet\base.py", line 1296, in startRunning
    ReactorBase.startRunning(cast(ReactorBase, self))
File "C:\Program Files\Python310\lib\site-packages\twisted\internet\base.py", line 840, in startRunning
    raise error.ReactorNotRestartable()
twisted.internet.error.ReactorNotRestartable

错误原因 #

Scrapy基于Twisted框架，其核心的反应器(reactor)是不可重启的——一旦调用process.start()启动并停止反应器后，同一个进程内无法再次启动它。原代码循环中反复创建CrawlerProcess并调用start()，第二次循环时反应器已被耗尽，因此触发错误。

解决方案 #

方案1：一次性调度所有批次（推荐） #

无需循环创建CrawlerProcess，一次性将所有URL分批，把每批任务调度到同一个进程中，只启动一次反应器即可完成所有任务。

修改后代码：

from scrapy.crawler import CrawlerProcess
from scrapy.utils.project import get_project_settings

# 提取所有待处理URL
unprocessed_urls = [item['crawl_url'] for item in crawler_table.find(crawl_timestamp=None)]
batch_size = 10
settings = get_project_settings()
process = CrawlerProcess(settings)

# 分批创建爬虫任务
for i in range(0, len(unprocessed_urls), batch_size):
    batch_urls = unprocessed_urls[i:i+batch_size]
    process.crawl(MySpider, start_urls=batch_urls)

# 启动一次反应器，处理所有任务
process.start()

方案2：子进程独立处理每批任务 #

如果必须分批启动独立进程，可使用subprocess模块为每批URL启动单独的爬虫进程，每个进程拥有独立的反应器，避免冲突。

主调度代码：

import subprocess
import sys

unprocessed_urls = [item['crawl_url'] for item in crawler_table.find(crawl_timestamp=None)]
batch_size = 10

for i in range(0, len(unprocessed_urls), batch_size):
    batch_urls = unprocessed_urls[i:i+batch_size]
    # 将批次URL作为参数传递给独立爬虫脚本
    subprocess.run([
        sys.executable, 'spider_batch.py',
        '--batch-urls', ','.join(batch_urls)
    ])

对应的独立爬虫脚本spider_batch.py：

import sys
from scrapy.crawler import CrawlerProcess
from scrapy.utils.project import get_project_settings

class MySpider(scrapy.Spider):
    name = 'my_spider'
    def __init__(self, start_urls=None, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if start_urls:
            self.start_urls = start_urls.split(',')

if __name__ == '__main__':
    settings = get_project_settings()
    process = CrawlerProcess(settings)
    # 解析传入的批次URL
    batch_urls = sys.argv[sys.argv.index('--batch-urls') + 1]
    process.crawl(MySpider, start_urls=batch_urls)
    process.start()

方案3：使用CrawlerRunner手动控制反应器 #

CrawlerRunner不会自动启动反应器，可手动控制任务调度和反应器生命周期，适合需要在爬虫完成后执行后续操作的场景。

示例代码：

from scrapy.crawler import CrawlerRunner
from scrapy.utils.project import get_project_settings
from twisted.internet import reactor

unprocessed_urls = [item['crawl_url'] for item in crawler_table.find(crawl_timestamp=None)]
batch_size = 10
settings = get_project_settings()
runner = CrawlerRunner(settings)

# 定义单批次爬取函数
def crawl_batch(batch_urls):
    return runner.crawl(MySpider, start_urls=batch_urls)

# 依次调度所有批次任务
deferred = crawl_batch(unprocessed_urls[0:batch_size])
for i in range(batch_size, len(unprocessed_urls), batch_size):
    batch_urls = unprocessed_urls[i:i+batch_size]
    deferred.addCallback(lambda _: crawl_batch(batch_urls))

# 所有任务完成后停止反应器
deferred.addCallback(lambda _: reactor.stop())

# 启动反应器
reactor.run()

关键注意事项 #

• 无论采用哪种方案，必须在URL处理完成后更新crawl_timestamp字段，否则后续批次会重复爬取已处理的URL。
• 方案1是最高效的实现方式，无需创建多进程，所有任务在同一个反应器中调度执行。

内容的提问来源于stack exchange，提问作者Roman