如何判断CrawlerProcess层级是否触发CloseSpider并终止爬虫循环

问题背景 #

循环运行Scrapy爬虫时，当爬虫触发CloseSpider异常，该异常被Twisted内部处理，无法在run_crawler()中捕获并向上传递，导致循环无法终止。

解决方案 #

核心思路是捕获Scrapy爬虫启动过程中Deferred对象的异常，将CloseSpider异常从Twisted的异步上下文传递到同步上下文，再通过多进程管道传递到主进程，最终终止循环。

关键修改点 #

1. 在run_crawler中捕获Deferred异常：

   • 获取爬虫启动的Deferred对象，添加Errback捕获CloseSpider异常
   • 捕获异常后停止Twisted reactor，并重新抛出异常让多进程捕获

2. 修正多进程异常传递逻辑：

   • 让自定义Process类传递异常对象和对应的traceback，而不是仅传递traceback字符串

修改后的完整代码 #

from __future__ import print_function
import multiprocessing as mp
import traceback
from time import sleep
from typing import Type
from scrapy import Spider
from scrapy.crawler import CrawlerProcess
from scrapy.exceptions import CloseSpider
from twisted.internet import reactor
from twisted.internet.defer import Deferred


class MyTestSpider(Spider):
    name = "my_test_spider"

    def __init__(self) -> None:
        raise CloseSpider("触发终止循环的异常")


class Process(mp.Process):
    def __init__(self, target: callable, *args, **kwargs):
        mp.Process.__init__(self, target=target, *args, **kwargs)
        self._pconn, self._cconn = mp.Pipe()
        self._exception = None

    def run(self):
        try:
            mp.Process.run(self)
            self._cconn.send((None, None))
        except Exception as e:
            tb = traceback.format_exc()
            self._cconn.send((e, tb))

    @property
    def exception(self):
        if self._pconn.poll():
            self._exception = self._pconn.recv()
        return self._exception


def run_crawler_loop(
    spider: Type[Spider],
    loop_wait_secs: int,
    **kwargs,
) -> None:
    while True:
        print("启动新一轮爬虫...")
        run_crawler_reactor_safe(spider=spider, **kwargs)
        sleep(loop_wait_secs)


def run_crawler_reactor_safe(spider: Type[Spider], **kwargs) -> None:
    process = Process(target=run_crawler, kwargs={"spider": spider} | kwargs)
    process.start()
    process.join()

    error, tb = process.exception
    if error:
        print(f"捕获到异常: {error}\n{tb}")
        # 这里可以添加邮件告警逻辑
        raise error  # 终止循环


def run_crawler(spider: Type[Spider], **kwargs) -> None:
    process = CrawlerProcess()
    crawler = process.create_crawler(spider)
    
    # 获取爬虫启动的Deferred对象
    d: Deferred = process.crawl(crawler_or_spidercls=crawler, **kwargs)
    
    # 添加Errback捕获CloseSpider异常
    def handle_error(failure):
        # 检查是否是CloseSpider异常
        if failure.check(CloseSpider):
            print(f"爬虫触发CloseSpider: {failure.value}")
            # 停止reactor
            reactor.stop()
            # 重新抛出异常，让多进程捕获
            raise failure.value
        return failure
    
    d.addErrback(handle_error)
    
    # 启动reactor，直到爬虫结束或异常触发
    reactor.run()


if __name__ == "__main__":
    try:
        run_crawler_loop(spider=MyTestSpider, loop_wait_secs=0)
    except CloseSpider:
        print("循环已终止")

代码说明 #

• Deferred异常捕获：通过d.addErrback(handle_error)监听爬虫启动过程中的异常，当捕获到CloseSpider时，停止reactor并重新抛出异常。
• 多进程异常传递：自定义Process类将异常对象和traceback一起通过管道发送到主进程，主进程run_crawler_reactor_safe接收后抛出异常，终止run_crawler_loop的循环。
• 循环终止逻辑：主进程捕获run_crawler_reactor_safe抛出的CloseSpider异常，最终终止整个循环。

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