Scrapy多次运行CrawlerProcess时触发ReactorNotRestartable问题

这个问题我之前在写Twisted相关测试的时候也碰到过，确实挺头疼的——Twisted的Reactor是个单例，一旦启动又停止后，内部的一些状态没法自动重置，直接再调用start()就会触发ReactorNotRestartable错误。好在有几种靠谱的方法能实现测试用例之间的隔离，让它们依次正常执行：

方法1：使用pytest-twisted插件自动管理Reactor #

这是最省心的方案，pytest-twisted插件会帮你在每个测试用例前后自动处理Reactor的启停和状态清理，完全不用手动操心。

步骤： #

1. 先安装插件：

pip install pytest-twisted

2. 改写你的测试用例，使用异步风格（配合Twisted的Deferred或者直接用async/await）：

from twisted.internet.defer import inlineCallbacks
import pytest

@pytest.inlineCallbacks
def test_first_run():
    # 调用你的run()方法
    yield run()
    # 这里添加你的断言逻辑
    assert ...

@pytest.inlineCallbacks
def test_second_run():
    yield run()
    assert ...

运行测试时，插件会自动为每个测试用例启动全新的Reactor环境，测试结束后自动停止并清理，彻底避免重启问题。

方法2：手动用PyTest Fixture实现Reactor重置 #

如果不想依赖第三方插件，可以自己写一个全局的PyTest Fixture，在每个测试用例前后手动处理Reactor的状态：

import pytest
from twisted.internet import reactor
from twisted.internet.defer import ensureDeferred, Deferred

@pytest.fixture(autouse=True, scope="function")
async def reset_reactor():
    # 测试前：确保Reactor处于停止状态
    if reactor.running:
        reactor.stop()
        # 等待Reactor完全停止
        stop_deferred = Deferred()
        reactor.addSystemEventTrigger("after", "shutdown", stop_deferred.callback, None)
        await stop_deferred
    
    # 运行测试用例
    yield
    
    # 测试后：停止Reactor并清理状态
    if reactor.running:
        reactor.stop()
        stop_deferred = Deferred()
        reactor.addSystemEventTrigger("after", "shutdown", stop_deferred.callback, None)
        await stop_deferred

# 测试用例写成异步风格
async def test_first_case():
    await ensureDeferred(run())
    assert ...

async def test_second_case():
    await ensureDeferred(run())
    assert ...

这个Fixture会自动作用于每个测试用例，确保每个测试用例启动前Reactor是干净停止的，测试结束后也会彻底停止Reactor。

方法3：切换到Twisted自带的Trial测试框架 #

如果项目允许切换测试框架，Twisted自带的twisted.trial会为每个测试用例创建独立的Reactor实例，天然支持测试隔离：

from twisted.trial.unittest import TestCase
from twisted.internet.defer import inlineCallbacks

class MyReactorTests(TestCase):
    @inlineCallbacks
    def test_first_run(self):
        yield run()
        self.assertEqual(...)
    
    @inlineCallbacks
    def test_second_run(self):
        yield run()
        self.assertEqual(...)

运行测试时直接用Trial命令：

trial your_test_module.py

Trial会自动处理Reactor的生命周期，每个测试用例都有完全独立的Reactor环境，根本不会碰到重启问题。

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