Scrapy 2.17 源码解析:ExecutionEngine 引擎调度循环详解

本文基于 Scrapy 2.17.0 + Twisted 26.4.0 的真实源码逐行分析。注意：Scrapy 从 2.14 开始对引擎做了 async/await 重构，本文讲的 start_async、_process_start_next、_Slot 在旧版本（≤ 2.13）中并不存在。网上大量教程仍停留在 _next_request_from_scheduler 时代，请注意版本差异。

前言

在之前的文章里，我们已经把 Scrapy 的下载器层啃得差不多了：

Scrapy：Downloader 下载器设计详解
Scrapy：DownloaderMiddlewareManager 设计详解
Scrapy：_RequestBodyProducer 类详解
Scrapy：DownloaderAwarePriorityQueue 队列设计详解

下载器解决的是「怎么把一个 Request 变成 Response」。但还有一个更上层的问题没回答：

谁来决定什么时候下载哪一个请求？下载完的结果又交给谁？并发到底卡在哪一层？

答案是 ExecutionEngine——Scrapy 的中枢。这篇我们把它拆开。

一、引擎在架构中的位置

先建立全局认知。Scrapy 的数据流是一个闭环：

引擎自己不做任何业务，它只干三件事：

从 Scheduler 取请求，丢给 Downloader
把 Downloader 的产物分流：Response → Scraper，Request → 重新入队
判断「什么时候该停下来」

看懂这三件事，引擎就通了。

二、`init`：三大组件的装配顺序

先看构造函数的关键片段（scrapy/core/engine.py）：

class ExecutionEngine:    _SLOT_HEARTBEAT_INTERVAL: float = 5.0    def __init__(self, crawler, spider_closed_callback):        self.crawler = crawler        self.settings = crawler.settings        self.signals = crawler.signals        self.logformatter = crawler.logformatter        self._slot: _Slot | None = None        self.spider: Spider | None = None        self.running: bool = False        self._starting: bool = False        self._stopping: bool = False        self.paused: bool = False        self.start_time: float | None = None        self._start: AsyncIterator[Any] | None = None        self._closewait: Deferred[None] | None = None        downloader_cls = load_object(self.settings["DOWNLOADER"])        try:            self.scheduler_cls = self._get_scheduler_class(crawler.settings)            self.downloader = downloader_cls(crawler)            self.scraper = Scraper(crawler)        except Exception:            if hasattr(self, "downloader"):                self.downloader.close()            raise

有三个细节值得停下来看。

1. `scheduler_cls` 只存类，不实例化

self.scheduler_cls = self._get_scheduler_class(crawler.settings)

注意存的是类而不是实例。因为调度器是 per-spider 的资源（它可能对应一个磁盘队列目录），必须等 open_spider_async() 时才能建。而下载器和 Scraper 是 per-engine 的，构造函数里直接建好。

2. 调度器做了接口校验

def _get_scheduler_class(self, settings: BaseSettings) -> type[BaseScheduler]:    scheduler_cls = load_object(settings["SCHEDULER"])    if not issubclass(scheduler_cls, BaseScheduler):        raise TypeError(            f"The provided scheduler class ({settings['SCHEDULER']})"            " does not fully implement the scheduler interface"        )    return scheduler_cls

自定义 SCHEDULER 时必须继承 BaseScheduler，鸭子类型在这里行不通。这是 Scrapy 少见的强制继承点之一。

3. `try/except` 里的资源清理

except Exception:    if hasattr(self, "downloader"):        self.downloader.close()    raise

如果 Scraper(crawler) 构造失败（比如某个 Pipeline 的 from_crawler 抛异常），已经建好的 downloader 必须关掉，否则它持有的 reactor 资源会泄漏。用 hasattr 判断，是因为异常可能发生在 downloader 赋值之前。

这是个很典型的部分构造失败处理范式，写自己的组件管理器时可以直接抄。

三、`_Slot`：引擎的运行时状态容器

class _Slot:    def __init__(self, close_if_idle, nextcall, scheduler):        self.closing: Deferred[None] | None = None        self.inprogress: set[Request] = set()        self.close_if_idle: bool = close_if_idle        self.nextcall: CallLaterOnce[None] = nextcall        self.scheduler: BaseScheduler = scheduler        self.heartbeat = create_looping_call(nextcall.schedule)    def add_request(self, request: Request) -> None:        self.inprogress.add(request)    def remove_request(self, request: Request) -> None:        self.inprogress.remove(request)        self._maybe_fire_closing()    async def close(self) -> None:        self.closing = Deferred()        self._maybe_fire_closing()        await maybe_deferred_to_future(self.closing)    def _maybe_fire_closing(self) -> None:        if self.closing is not None and not self.inprogress:            if self.nextcall:                self.nextcall.cancel()                if self.heartbeat.running:                    self.heartbeat.stop()            self.closing.callback(None)

_Slot 是 per-spider 的运行时状态。字段职责：

字段	类型	作用
`inprogress`	`set[Request]`	当前「已从调度器取出、尚未处理完」的请求
`scheduler`	`BaseScheduler`	本次爬取的调度器实例
`nextcall`	`CallLaterOnce`	调度循环的触发器（下一节详解）
`heartbeat`	`LoopingCall`	5 秒一次的兜底心跳
`closing`	`Deferred` 或 `None`	关闭信号量
`close_if_idle`	`bool`	空闲时是否自动关闭

优雅关闭的两阶段协议

close() 的写法很讲究：

async def close(self) -> None:    self.closing = Deferred()      # 阶段一：立起「正在关闭」标志    self._maybe_fire_closing()     # 阶段二：立即检查一次（可能已经空了）    await maybe_deferred_to_future(self.closing)

设置 self.closing 之后立刻调一次 _maybe_fire_closing()，是为了处理「调用 close() 时 inprogress 已经是空」的场景——否则就会永久挂起，因为再也不会有 remove_request() 来触发回调了。

而正常情况下，每个请求处理完都会走 remove_request() → _maybe_fire_closing()，直到最后一个请求归零时才真正 fire。

这是「引用计数式关闭」的标准实现：设标志 → 立即自检 → 每次减引用时复检。

`heartbeat` 为什么存在

self.heartbeat = create_looping_call(nextcall.schedule)# ...self._slot.heartbeat.start(self._SLOT_HEARTBEAT_INTERVAL)   # 5.0 秒

源码注释说得很清楚：

a periodic call to that processing method for scenarios where the scheduler reports having pending requests but returns none.

翻译过来：调度器可能会「撒谎」。has_pending_requests() 返回 True，但 next_request() 返回 None。

最典型的场景是分布式调度器（比如 scrapy-redis）：Redis 队列里确实标记有任务，但当前这个节点抢不到。如果没有心跳，引擎会因为「没有任何事件触发 nextcall」而永久卡死。5 秒心跳就是这个死锁的保险丝。

四、`CallLaterOnce`：调度循环的去重开关

这是整个引擎最精妙的一个小类，在 scrapy/utils/reactor.py：

class CallLaterOnce(Generic[_T]):    """Schedule a function to be called in the next reactor loop, but only if    it hasn't been already scheduled since the last time it ran.    """    def __init__(self, func, *a, **kw):        self._func = func        self._a = a        self._kw = kw        self._call: CallLaterResult | None = None        self._deferreds: list[Deferred[None]] = []    def schedule(self, delay: float = 0) -> None:        from scrapy.utils.asyncio import call_later        if self._call is None:            # 关键：已排期就不再排            self._call = call_later(delay, self)    def cancel(self) -> None:        if self._call:            self._call.cancel()    def __call__(self) -> _T:        from scrapy.utils.asyncio import call_later        self._call = None                 # 先清空，允许重新排期        result = self._func(*self._a, **self._kw)        for d in self._deferreds:            call_later(0, d.callback, None)        self._deferreds.clear()        return result

它解决什么问题

引擎里到处都在喊「该干活了」：

请求入队后 → nextcall.schedule()
下载完成后 → nextcall.schedule()
请求从 slot 移除后 → nextcall.schedule()
心跳每 5 秒 → nextcall.schedule()

如果每次都真的排一个 reactor 回调，那么一次下载完成可能触发三四次重复调度，并发 16 的情况下每轮就是几十次无效调用。

CallLaterOnce 用一个 self._call 指针做幂等：

schedule()
时如果 _call 非空，说明已经排过了，直接丢弃
真正执行时先把 _call = None，重新开放排期

于是无论一个 reactor tick 内被 schedule() 多少次，_start_scheduled_requests()只会执行一次。

一句话概括：这是 reactor 层面的「合并调用」（coalescing），和前端 requestAnimationFrame 去抖是同一个思想。

五、启动链路：`start_async`

async def start_async(self, *, _start_request_processing: bool = True) -> None:    if self._starting:        raise RuntimeError("Engine already running")    self.start_time = time()    self._starting = True    await self.signals.send_catch_log_async(signal=signals.engine_started)    if self._stopping:        return    if _start_request_processing and self.spider is None:        # require an opened spider when not run in scrapy shell        return    self.running = True    self._closewait = Deferred()    if _start_request_processing:        coro = self._start_request_processing()        if is_asyncio_available():            self._start_request_processing_awaitable = asyncio.ensure_future(coro)        else:            self._start_request_processing_awaitable = Deferred.fromCoroutine(coro)    with contextlib.suppress(asyncio.exceptions.CancelledError):        await maybe_deferred_to_future(self._closewait)

三个要点：

1. _closewait 是引擎主体的生命线

start_async() 最后 await self._closewait，这个 Deferred 只在 stop_async() 里被 callback。也就是说 start_async() 会一直挂着，直到引擎被显式停止。整个爬虫的生命周期就是这个 await 的时长。

2. _start_request_processing 是后台任务，不是被 await 的

self._start_request_processing_awaitable = asyncio.ensure_future(coro)

注意它用 ensure_future甩到后台，而不是 await——因为主协程要去 await _closewait。这里保存句柄，是为了 stop_async() 时能 .cancel() 它。

3. is_asyncio_available() 的分叉

asyncio reactor 下用 asyncio.ensure_future，否则用 Deferred.fromCoroutine。源码注释解释了为什么不统一包成 Deferred：

not wrapping in a Deferred here to avoid https://github.com/twisted/twisted/issues/12470

这是在绕 Twisted 的一个已知 bug（取消时的行为异常）。属于典型的「源码里的历史包袱」，看不懂时优先去翻 issue。

六、双循环模型

这是 2.14 重构后最大的结构变化：引擎里其实跑着两个独立的循环。

循环 A：start 迭代循环（消费 `Spider.start()`）

async def _start_request_processing(self) -> None:    try:        self._slot.nextcall.schedule()        self._slot.heartbeat.start(self._SLOT_HEARTBEAT_INTERVAL)        while self._start and self.spider and self.running:            await self._process_start_next()            if not self.needs_backout():                self._slot.nextcall.schedule()                await self._slot.nextcall.wait()    except (asyncio.exceptions.CancelledError, CancelledError):        return    except Exception:        self._start_request_processing_awaitable = None        logger.error("Error while processing requests from start()",                     exc_info=True, extra={"spider": self.spider})        await self.stop_async()

async def _process_start_next(self) -> None:    assert self._start is not None    try:        item_or_request = await anext(self._start)    except StopAsyncIteration:        self._start = None    except Exception as exception:        self._start = None        logger.error(f"Error while reading start items and requests: {exception}.\n"                     f"{format_exc()}", exc_info=True)    else:        if not self.spider:            return        if isinstance(item_or_request, Request):            self.crawl(item_or_request)        else:            _schedule_coro(                self.scraper.start_itemproc_async(item_or_request, response=None)            )            self._slot.nextcall.schedule()

关键设计：Spider.start() 现在是一个 AsyncIterator，被「一次拉一个」地消费。

while 循环里那句 await self._slot.nextcall.wait() 是背压的核心——只有当调度循环真的跑过一轮之后，才继续从 start() 里拉下一个。

这解决了老版本的一个经典问题：当 start_requests 是个生成 100 万个 URL 的生成器时，老版本会疯狂往调度器里灌，内存直接爆掉。现在是严格的拉模式：下游消化一个，上游才生产一个。

顺带一提：_process_start_next() 里 Item 和 Request 是分流的——start() 现在可以直接 yield Item，绕过下载器直接进 Pipeline。这是 2.13+ 的新能力。

循环 B：调度循环（消费 `Scheduler`）

def _start_scheduled_requests(self) -> None:    if self._slot is None or self._slot.closing is not None or self.paused:        return    while not self.needs_backout():        if not self._start_scheduled_request():            break    if self.spider_is_idle() and self._slot.close_if_idle:        self._spider_idle()

这就是被 CallLaterOnce 包着的那个函数。它是同步的，一口气把调度器榨干，直到：

needs_backout()
为真（并发满了），或
_start_scheduled_request()
返回 False（调度器空了）

然后顺手检查一次空闲状态。

两个循环通过 nextcall 握手：循环 A 每拉一个就 schedule() + wait()；循环 B 跑完后，通过 CallLaterOnce.__call__ 里的 _deferreds 唤醒循环 A。

七、单个请求的完整生命周期

def _start_scheduled_request(self) -> bool:    request = self._slot.scheduler.next_request()    if request is None:        self.signals.send_catch_log(signals.scheduler_empty)        return False    d: Deferred[Response | Request] = self._download(request)    d.addBoth(self._handle_downloader_output, request)    d.addErrback(lambda f: logger.info(        "Error while handling downloader output",        exc_info=failure_to_exc_info(f), extra={"spider": self.spider}))    d2: Deferred[None] = d.addBoth(partial(self._remove_request, request=request))    d2.addErrback(lambda f: logger.info(        "Error while removing request from slot",        exc_info=failure_to_exc_info(f), extra={"spider": self.spider}))    slot = self._slot    d2.addBoth(lambda _: slot.nextcall.schedule())    d2.addErrback(lambda f: logger.info(        "Error while scheduling new request",        exc_info=failure_to_exc_info(f), extra={"spider": self.spider}))    return True

这条 Deferred 链是教科书级的错误隔离写法。拆解：

步骤	回调	类型	作用
1	`self._download(request)`	起点	下载
2	`_handle_downloader_output`	`addBoth`	分流 Response / Request
3	`lambda f: logger.info(...)`	`addErrback`	隔离步骤 2 的异常
4	`_remove_request`	`addBoth`	从 `inprogress` 移除
5	`lambda f: logger.info(...)`	`addErrback`	隔离步骤 4 的异常
6	`slot.nextcall.schedule()`	`addBoth`	触发下一轮调度
7	`lambda f: logger.info(...)`	`addErrback`	隔离步骤 6 的异常

核心模式：每个 addBoth 后面都紧跟一个 addErrback。

为什么？因为在 Twisted 里，如果某个回调抛异常而没人接住，Failure 会一路传到链尾，导致后续的清理动作（移除请求、触发调度）全部被跳过。那样这个请求就会永远卡在 inprogress 里，引擎再也不会 idle，爬虫永远不退出。

用 addBoth + addErrback 交替，等价于给每一步套一个 try/except: log，保证清理逻辑一定执行。

另一个细节：slot = self._slot 先存成局部变量，再在 lambda 里用。因为回调触发时 self._slot 可能已经被置 None（spider 关了），闭包捕获局部变量才安全。这是异步代码里非常容易踩的坑。

八、`needs_backout`：三级背压

def needs_backout(self) -> bool:    return (        not self.running        or not self._slot        or bool(self._slot.closing)        or self.downloader.needs_backout()        or self.scraper.slot.needs_backout()    )

背压来自三个独立的闸门。

闸门 1：下载器并发

# scrapy/core/downloader/__init__.pydef needs_backout(self) -> bool:    return len(self.active) >= self.total_concurrency

total_concurrency = CONCURRENT_REQUESTS（默认 16）。这是请求条数维度。

闸门 2：Scraper 内存水位

# scrapy/core/scraper.pydef needs_backout(self) -> bool:    return self.active_size > self.max_active_size

max_active_size = SCRAPER_SLOT_MAX_ACTIVE_SIZE（默认 5,000,000 字节，约 5 MB）。

注意它统计的是字节数而不是条数：

MIN_RESPONSE_SIZE = 1024def add_response_request(self, result, request):    ...    if isinstance(result, Response):        self.active_size += max(len(result.body), self.MIN_RESPONSE_SIZE)    else:        self.active_size += self.MIN_RESPONSE_SIZE

每个响应按 max(body长度, 1024) 计入。这意味着：

爬大文件时，几个响应就能顶满 5 MB，并发被迫降下来
爬小页面时，1 KB 的地板价保证条数不会失控

这是一个基于内存而非计数的背压设计，比单纯限制条数聪明得多。 解析速度跟不上下载速度时，它会自动踩刹车。

闸门 3：引擎状态

not self.running 与 self._slot.closing——停止过程中不再放新请求出去。

对比总结

闸门	配置项	默认值	维度
下载器并发	`CONCURRENT_REQUESTS`	16	请求条数
Scraper 水位	`SCRAPER_SLOT_MAX_ACTIVE_SIZE`	5 MB	响应字节数
Item 并发	`CONCURRENT_ITEMS`	100	Item 条数

调优时的判断顺序：先看 len(active) 是否顶到 16（下载器瓶颈），再看内存是否卡在 5 MB（解析瓶颈）。两者的优化手段完全不同。

九、`_handle_downloader_output`：重定向为什么不占并发

@inlineCallbacksdef _handle_downloader_output(self, result, request):    if not isinstance(result, (Request, Response, Failure)):        raise TypeError(            f"Incorrect type: expected Request, Response or Failure, "            f"got {type(result)}: {result!r}")    # downloader middleware can return requests (for example, redirects)    if isinstance(result, Request):        self.crawl(result)        return    try:        yield self.scraper.enqueue_scrape(result, request)    except Exception:        logger.error("Error while enqueuing scrape",                     exc_info=True, extra={"spider": self.spider})

短短十几行，藏着一个常被误解的机制：

下载器中间件返回 Request 时（最典型的就是 RedirectMiddleware 处理 302），这个新请求不是「接着下载」，而是被 self.crawl() 重新扔回调度器。

后果是：

重定向后的 URL 会重新过一遍 dupefilter（所以重定向目标被去重是正常现象）
它会重新排队，而不是插队
原请求的 Deferred 链在这里就结束了，_remove_request 立刻执行，并发槽位马上释放

很多人以为「一次 302 等于占用两个并发」，其实不是。302 的两跳在引擎眼里是两个完全独立的请求，串行占用同一个槽位。

十、空闲判定与关闭

`spider_is_idle`：四个条件全为真才算空闲

def spider_is_idle(self) -> bool:    if self._slot is None:        raise RuntimeError("Engine slot not assigned")    if not self.scraper.slot.is_idle():        # 1. Scraper 还有活        return False    if self.downloader.active:                 # 2. 下载器还有活        return False    if self._start is not None:                # 3. start() 还没迭代完        return False    return not self._slot.scheduler.has_pending_requests()   # 4. 调度器还有货

注意条件 3：只要 Spider.start() 这个异步迭代器还没抛 StopAsyncIteration，引擎就永远不算空闲。

这解释了一个常见困惑：

为什么 start() 里写了个 while True 死循环，爬虫就永远不会自动结束？

因为 self._start 永远不是 None，条件 3 永远为 False。

`_spider_idle`：可以被信号「救回来」

def _spider_idle(self) -> None:    expected_ex = (DontCloseSpider, CloseSpider)    res = self.signals.send_catch_log(        signals.spider_idle, spider=self.spider, dont_log=expected_ex)    detected_ex = {        ex: x.value        for _, x in res        for ex in expected_ex        if isinstance(x, Failure) and isinstance(x.value, ex)    }    if DontCloseSpider in detected_ex:        return    if self.spider_is_idle():        ex = detected_ex.get(CloseSpider, CloseSpider(reason="finished"))        _schedule_coro(self.close_spider_async(reason=ex.reason))

这是 Scrapy 扩展性最强的钩子之一：

任何 spider_idle 信号的接收者抛 DontCloseSpider → 引擎放弃本次关闭
抛 CloseSpider(reason=...) → 用自定义原因关闭

scrapy-redis 的「永不退出」就是靠前者实现的：监听 spider_idle，每次都抛 DontCloseSpider，同时去 Redis 拉新任务。

注意 if self.spider_is_idle() 这个二次确认——因为信号处理器在执行期间可能已经往调度器里塞了新请求，此时就不该关了。

十一、2.14 到 2.17 的 async 化对照表

如果你在维护老代码，这张表很重要：

旧 API（已废弃）	新 API	说明
`engine.start()`	`engine.start_async()`	返回协程而非 Deferred
`engine.stop()`	`engine.stop_async()`	同上
`engine.close()`	`engine.close_async()`	同上
`engine.open_spider(spider)`	`engine.open_spider_async()`	spider 参数取消
`engine.close_spider(spider, reason)`	`engine.close_spider_async(reason=...)`	spider 参数取消
`engine.download(request)`	`engine.download_async(request)`	返回协程
`Spider.start_requests()`	`Spider.start()`	变成 async generator

旧 API 都还在，但会发 ScrapyDeprecationWarning：

def start(self, _start_request_processing: bool = True) -> Deferred[None]:    warnings.warn(        "ExecutionEngine.start() is deprecated, use start_async() instead",        ScrapyDeprecationWarning, stacklevel=2)    return deferred_from_coro(        self.start_async(_start_request_processing=_start_request_processing))

统一的兼容手法：旧同步方法 = deferred_from_coro(新协程方法)。想给自己的库做 async 迁移，这个模式可以直接照搬。

十二、动手：观测引擎的实时状态

光看源码不过瘾，写个扩展把引擎内部状态打出来。

# engine_monitor.pyimport loggingfrom scrapy import signalsfrom scrapy.exceptions import NotConfiguredlogger = logging.getLogger(__name__)class EngineMonitor:    """周期性打印 ExecutionEngine 的内部状态"""    def __init__(self, crawler, interval: float):        self.crawler = crawler        self.interval = interval        self.task = None    @classmethod    def from_crawler(cls, crawler):        interval = crawler.settings.getfloat("ENGINE_MONITOR_INTERVAL", 5.0)        if not interval:            raise NotConfigured        ext = cls(crawler, interval)        crawler.signals.connect(ext.spider_opened, signal=signals.spider_opened)        crawler.signals.connect(ext.spider_closed, signal=signals.spider_closed)        return ext    def spider_opened(self, spider):        from twisted.internet import task        self.task = task.LoopingCall(self.report)        self.task.start(self.interval, now=False)    def spider_closed(self, spider):        if self.task and self.task.running:            self.task.stop()    def report(self):        engine = self.crawler.engine        slot = engine._slot        if slot is None:            return        downloader = engine.downloader        scraper_slot = engine.scraper.slot        logger.info(            "[engine] inprogress=%d | downloader.active=%d/%d | "            "scraper.active_size=%d/%d | scheduler.pending=%s | "            "backout=%s | idle=%s",            len(slot.inprogress),            len(downloader.active), downloader.total_concurrency,            scraper_slot.active_size, scraper_slot.max_active_size,            slot.scheduler.has_pending_requests(),            engine.needs_backout(),            engine.spider_is_idle(),        )

启用：

# settings.pyEXTENSIONS = {    "engine_monitor.EngineMonitor": 500,}ENGINE_MONITOR_INTERVAL = 3.0

跑一个真实爬虫，日志大概长这样：

[engine] inprogress=16 | downloader.active=16/16 | scraper.active_size=1245184/5000000 | scheduler.pending=True | backout=True | idle=False[engine] inprogress=16 | downloader.active=16/16 | scraper.active_size=3891200/5000000 | scheduler.pending=True | backout=True | idle=False[engine] inprogress=9  | downloader.active=9/16  | scraper.active_size=5242880/5000000 | scheduler.pending=True | backout=True | idle=False

看第三行：downloader.active 只有 9（没顶满 16），但 backout=True——瓶颈在 Scraper 的内存水位，不在下载并发。

这时候盲目调大 CONCURRENT_REQUESTS 完全没用，正确的做法是：

优化 Pipeline / 解析逻辑（让 active_size 降得更快），或
调大 SCRAPER_SLOT_MAX_ACTIVE_SIZE（用内存换吞吐）

这就是读引擎源码的实际价值——性能调优不再靠猜。

十三、几个容易踩的坑

坑 1：在 `spider_idle` 里同步塞请求

# 错误示范def spider_idle(self, spider):    for url in fetch_new_urls():        # 阻塞 I/O        spider.crawler.engine.crawl(Request(url))    raise DontCloseSpider

spider_idle 是同步信号，阻塞 I/O 会卡死整个 reactor。正确做法是提前预取，或用 _schedule_coro 甩到后台。

坑 2：把 `engine._slot` 缓存成实例属性

_slot 在 close_spider_async() 里会被置 None，缓存下来的引用会变成悬垂指针。要用就每次现取，或者像源码那样在闭包里存局部变量。

坑 3：以为 `CONCURRENT_REQUESTS` 是唯一的并发开关

前面说过，SCRAPER_SLOT_MAX_ACTIVE_SIZE 和 CONCURRENT_ITEMS 同样会成为瓶颈。三个都要看。

坑 4：`start()` 里 yield 太快

新版是拉模式，但如果你在 start() 里写：

async def start(self):    for url in self.urls:      # 一百万个        yield Request(url)

调度器仍会把它们全接下来（内存队列）。真正的解法是配 SCHEDULER_DISK_QUEUE 落盘，或者干脆在 start() 里做限流。

总结

ExecutionEngine 的设计可以浓缩成五句话：

引擎不做业务
，只负责在 Scheduler、Downloader、Scraper 之间搬运和分流
双循环
：start 迭代循环（拉模式消费 Spider.start()）+ 调度循环（榨干 Scheduler）
CallLaterOnce 做调度去重
，把一个 tick 内的 N 次「该干活了」合并成 1 次
三级背压
：下载器按条数、Scraper 按字节、Item 按条数，任一顶格都会踩刹车
addBoth 与 addErrback 交替
保证清理逻辑不被异常跳过，这是引擎不卡死的根本

理解引擎之后，再回头看 CONCURRENT_REQUESTS 调不动的问题、爬虫不退出的问题、302 占并发的问题，答案都在源码里写着。

参考

Scrapy 源码：scrapy/core/engine.py
Scrapy 源码：scrapy/utils/reactor.py
Scrapy 源码：scrapy/core/scraper.py
官方文档：Architecture overview
官方文档：Settings reference

下一篇：《知识拓展：CallLaterOnce —— Scrapy 调度循环的去重开关》，我们把本文第四节那个小类单独拎出来，从 reactor 事件模型讲到它与 asyncio.call_soon、前端 requestAnimationFrame 去抖的同构性，并手写一个等价实现。

如果这篇对你有帮助，点赞收藏是最大的支持。

前言

一、引擎在架构中的位置

二、__init__：三大组件的装配顺序

1. scheduler_cls 只存类，不实例化

2. 调度器做了接口校验

3. try/except 里的资源清理

三、_Slot：引擎的运行时状态容器

优雅关闭的两阶段协议

heartbeat 为什么存在

四、CallLaterOnce：调度循环的去重开关