2.2 排班的艺术：抓取策略与调度

2.2 排班的艺术：抓取策略与调度

本节摘要：调度器的核心是 URL 边界队列——抓谁、先抓谁、多久抓一次。本节对比广度优先、深度优先、优先级调度、增量调度四种排班方式，给出频控与礼貌间隔的落码实现，并用 Crawl4AI 的深度爬取策略与并发分发器完成一次带过滤器的站点级排班。

排班问题的本质

把调度器想象成车间里那块排班板：板上钉着待抓的 URL 卡片，每天要回答三个问题——先抓谁（顺序策略）、一次派几辆车（并发度）、对同一个站点多久来一次（礼貌间隔）。三个问题对应三种失败模式：顺序错了，重要页面排在三万条后面，验收时还没抓到；并发错了，要么吞吐不足，要么把目标站压出故障；间隔错了，轻则被封，重则触犯 1.5 节的纪律红线。

排班策略的谱系不复杂。广度优先先抓完浅层再深入，适合"摸清站点结构"与"快速铺开覆盖"；深度优先沿一条链接链走到黑，适合话题追踪与路径明确的采集；优先级调度给 URL 打分排序，列表页高于详情页、更新频繁的栏目高于静态页；增量调度则换了一种问题框架——不是"抓一遍"而是"持续抓"，靠内容指纹判断变化。四者不互斥，真实排班板常常是组合：广度发现、优先级排序、增量跟进。

图 2-2 四种排班策略的适用地图

手写排班板：优先级加礼貌频控

先不借助框架，把排班板的最小逻辑写出来。优先级用堆实现，礼貌频控沿用 1.5 节的按域名间隔思路：

import heapq, time, collections

class Scheduler:
    """最小排班板：优先级出队 + 按域名礼貌间隔"""
    def __init__(self, min_interval: float = 2.0):
        self.heap: list = []                          # 元素：(负优先级, 序号, url)
        self.counter = 0                              # 序号防止优先级相同时比较 URL 报错
        self.last_seen: dict[str, float] = collections.defaultdict(float)
        self.min_interval = min_interval
        self.seen: set[str] = set()                   # URL 去重，工位五前置到入队时

    def push(self, url: str, priority: int = 0) -> bool:
        """入队；重复 URL 拒收，返回是否入队成功"""
        if url in self.seen:
            return False
        self.seen.add(url)
        heapq.heappush(self.heap, (-priority, self.counter, url))
        self.counter += 1
        return True

    def pop(self) -> tuple[str, float] | None:
        """出队并等待礼貌间隔，返回 (url, 等待秒数)"""
        if not self.heap:
            return None
        _, _, url = heapq.heappop(self.heap)
        domain = url.split("/")[2]
        elapsed = time.monotonic() - self.last_seen[domain]
        wait = max(0.0, self.min_interval - elapsed)
        time.sleep(wait)
        self.last_seen[domain] = time.monotonic()
        return url, wait

board = Scheduler(min_interval=1.0)
board.push("https://news.example/list", priority=10)   # 列表页高分
board.push("https://news.example/post/9", priority=3)  # 详情页低分
board.push("https://news.example/list", priority=10)   # 重复入队被拒
url, w = board.pop()
print(url, w)   # 输出：https://news.example/list 0.0（最高分先出，首次无等待）
​

三十行代码覆盖了排班板的三要素：顺序（堆）、去重（入队时拒收）、频控（按域名间隔）。它的局限也同样明显——单线程、无并发、没有深度控制，这正是框架要接管的部分。

交给引擎：深度爬取与并发分发

任务单升级到"归档某社区近三年全部帖子"（1.2 节的深度爬取形态），排班要加两个件：深度上限与链接过滤器。Crawl4AI 把这两件做成了声明式配置：

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig, CacheMode
from crawl4ai import BFSDeepCrawlStrategy
from crawl4ai.content_scraping_strategy import PruningContentFilter
from crawl4ai.deep_crawling.filters import DomainFilter, URLPatternFilter

async def deep_archive(start_url: str, max_pages: int = 20):
    run_cfg = CrawlerRunConfig(
        deep_crawl_strategy=BFSDeepCrawlStrategy(
            max_depth=2,                 # 深度上限：首页起两层，够覆盖列表到详情
            max_pages=max_pages,         # 页数熔断：保护预算与目标站
            include_external=False,      # 不越出目标域名，纪律与效率双保险
        ),
        cache_mode=CacheMode.BYPASS,
        content_filter=PruningContentFilter(),   # 分拣台预告：正文净化，噪声先剪掉
    )
    async with AsyncWebCrawler(config=BrowserConfig(headless=True)) as crawler:
        results = await crawler.arun(start_url, config=run_cfg)
        print("抓取页数:", len(results))            # 输出：抓取页数: 20（触顶熔断）
        print("首页深度:", results[0].metadata.get("depth"))
        # 输出：首页深度: 0（起点页深度为0，列表为1，详情为2）
        for r in results[:3]:
            print(r.url, "->", len(r.markdown.raw_markdown), "字符")
        # 输出：起始页与两层级页面各自的 Markdown 长度

asyncio.run(deep_archive("https://community.example.com"))
​

配置里有三个值得咀嚼的决策。max_depth=2 对应"列表到详情"的最短路径——深度每加一，页数近似翻倍，没有验收依据不放开。include_external=False 同时是效率与纪律约束，越出域名既浪费预算也扩大合规审查面。URLPatternFilter 这类过滤器可以再按路径模式筛选（比如只进帖子路径），让边界队列只收想要的卡片。

并发侧，arun_many 配合内存自适应分发器，把"一次派几辆车"变成资源约束下的自动决策：

from crawl4ai.async_dispatcher import MemoryAdaptiveDispatcher

async def batch_fetch(urls: list[str]):
    dispatcher = MemoryAdaptiveDispatcher(
        memory_threshold_percent=70.0,   # 内存水位超过70%就收紧并发
        max_session_permit=8,            # 同时最多8个会话，礼貌与吞吐的平衡点
    )
    run_cfg = CrawlerRunConfig(cache_mode=CacheMode.BYPASS, page_timeout=30000)
    async with AsyncWebCrawler(config=BrowserConfig(headless=True)) as crawler:
        results = await crawler.arun_many(urls, config=run_cfg, dispatcher=dispatcher)
        ok = sum(1 for r in results if r.success)
        print(f"成功 {ok}/{len(urls)}")       # 输出：成功 8/8（示例任务全绿）
        return results

asyncio.run(batch_fetch([f"https://community.example.com/t/{i}" for i in range(8)]))
​

常见坑：并发度开到几十，礼貌频控就名存实亡。max_session_permit 的上限应该由目标站点的承载与你的纪律承诺倒推，而不是由本机内存正推——内存自适应保护的是你的机器，不是对方的服务。

关键直觉：排班的收益曲线是先陡后平的。从串行到小并发（4 到 8）吞吐近乎翻倍；从 16 到 64，吞吐增幅常常不到两成，封禁风险却陡增。调度室的老经验：并发度停在曲线拐点，不要停在机器极限。