8. Funboost 用于爬虫

funboost 的分布式函数调度能力，自动给爬虫函数加30多种控制功能，一键拥有多种并发模式，自带可视化监控和告警，以及高级的熔断保护等功能，远强于普通爬虫框架的任务调度能力。所以自动天生非常适合自由写爬虫。 可以 funboost + (funspider 或 boost_spider 或 boost_scrapy 或 your_utils(你自己打磨积累的任何工具类)) 任意自由混合搭配

funboost是万能函数调度框架，可以直接用于爬虫， 也可以搭配 funboost爬虫全家桶(funboost爬虫全家桶包括funspider、boost_spider、boost_scrapy)。 其中 boost_scrapy 是boost_spider 项目下的1个副爬虫框架，专门模仿scrapy yield Request的语法，因为有些人scrapy中毒了，写爬虫不写yield 就不痛快。

8.1 核心理念：函数调度器 vs URL调度器

Funboost 是函数调度器，Scrapy 是URL调度器。这一句话道尽了二者的本质差异。

• Scrapy 调度的是一个 Request 对象，框架替用户发请求、回调解析、走 Pipeline 存储。用户被"框架生命周期"束缚。

• Funboost 调度的是一个完整的 Python 函数。框架只负责：何时调用、并发几个、失败怎么办。函数内部你写什么代码完全自由。

from funboost import boost, BoosterParams, BrokerEnum

@boost(BoosterParams(queue_name='crawl_detail', broker_kind=BrokerEnum.REDIS_ACK_ABLE, qps=5, max_retry_times=3))
def crawl_detail(url: str):
    # 这里你拥有 Python 的全部自由：
    # 用 requests/httpx/aiohttp 发请求、用 parsel/bs4 解析、用 pymongo/sqlalchemy 存储
    # 用 selenium/playwright 操作浏览器、调用你 utils/ 下所有现有函数
    import requests
    resp = requests.get(url, proxies=get_my_proxy(), headers=get_my_headers())
    data = parse_detail(resp.text)
    save_to_mysql(data)

# 发布任务
crawl_detail.push(url='https://example.com/article/123')
# 启动消费
crawl_detail.consume()

核心结论：Funboost 写爬虫就是"写函数就能爬虫"，加一个 @boost 装饰器，你的函数就自动获得了分布式、并发、重试、控频、断点续爬等 30+ 种能力。

---

8.2 @boost 装饰器赋予爬虫函数的 30+ 种控制能力

一个 @boost(BoosterParams(...)) 装饰器，让你的爬虫函数自动具备以下能力（无需一行额外代码）：

能力	BoosterParams 参数	说明
分布式消费	broker_kind	支持近 50 种消息队列（Redis、RabbitMQ、Kafka、SQLite、内存、PostgreSQL、NATS、gRPC 等）
并发模式	concurrent_mode	线程 / Gevent / Eventlet / Asyncio / 单线程
并发数量	concurrent_num	如 50 个线程并发消费
精准 QPS 限速	qps	精确到每秒 N 次请求，无视响应耗时
分布式全局控频	is_using_distributed_frequency_control	多台机器共享 QPS 上限
自动重试	max_retry_times	函数抛异常自动重试，含 HTTP 200 但内容错误的场景
高级重试策略	is_using_advanced_retry	指数退避、随机抖动
超时杀死	function_timeout	函数运行超时自动 kill
消费确认(ACK)	内置机制	崩溃/断电/重启消息不丢失
消息过期丢弃	msg_expire_seconds	丢弃过旧的堆积消息
任务去重(入参级)	do_task_filtering	基于函数入参自动去重，忽略 URL 中的噪音参数
去重有效期	task_filtering_expire_seconds	适合周期性更新爬取
死信队列	is_push_to_dlx_queue_when_retry_max_times	毒丸消息隔离
消费结果持久化	function_result_status_persistance_conf	自动记录成功/失败/耗时到 MongoDB
RPC 模式	is_using_rpc_mode	发布任务后同步等待并获取爬取结果
定时爬取	allow_run_time_cron	Cron 表达式定时触发
多进程叠加	multi_process_consume(N)	多线程/协程 × 多进程 × 多机器，榨干所有 CPU
远程部署	fabric_deploy(...)	一键部署爬虫到远程服务器
Web 管理	funboost web manager	实时监控消费速率、积压，查看消费结果和消费者状态
远程 Kill 任务	is_support_remote_kill_task	远程终止指定任务

对比 Scrapy：以上所有能力在 Scrapy 中要么不支持，要么需要复杂配置/开发中间件/安装插件/改源码才能实现。Funboost 只需一行装饰器全部搞定。

---

8.3 为什么 Funboost 不需要"插件"和"中间件"

8.3.1 整个 PyPI 都是 Funboost 的天然生态

Scrapy 需要 scrapy-redis、scrapy-selenium、scrapy-playwright 等专用插件，是因为框架内部封闭——只有写成 Middleware/Pipeline 形状的代码才能被调用。

Funboost 不需要任何插件。原因很简单：你的函数内部可以直接 import 任何 Python 包。

@boost(BoosterParams(queue_name='render_page', concurrent_num=5))
def render_and_crawl(url: str):
    from playwright.sync_api import sync_playwright  # 直接 import，无需插件
    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        page.goto(url)
        page.click('#load-more')
        page.wait_for_selector('.content')
        html = page.content()
        browser.close()
    # 直接用 parsel 解析
    from parsel import Selector
    sel = Selector(html)
    title = sel.css('h1::text').get()
    # 直接存 MongoDB
    from pymongo import MongoClient
    MongoClient().db.articles.insert_one({'url': url, 'title': title})

• 要换代理IP？直接写 proxies=random.choice(proxy_list) ——不需要 Downloader Middleware

• 要换 User-Agent？直接写 headers={'User-Agent': random_ua()} ——不需要 Spider Middleware

• 要存数据库？直接写 db.insert(data) ——不需要 Item Pipeline

• 要用 Playwright？直接 import playwright ——不需要 scrapy-playwright 插件

Funboost 的设计哲学：框架只管调度，不插手你怎么发请求、怎么解析、怎么存储。你项目里 utils/ 文件夹下积累多年的宝贵工具函数，可以零成本直接复用。

8.3.2 正面驳斥"scrapy 插件多 = scrapy 有优势"的误解

有人认为 scrapy 有 scrapy-redis、scrapy-selenium、scrapy-playwright 等大量插件，说明生态丰富是优势。这恰恰是本末倒置：

1. Scrapy 需要插件，正是其框架封闭的证据。如果你能在爬虫代码中直接 import playwright 使用，为什么还需要有人专门包一层 scrapy-playwright？正是因为 Scrapy 框架把 HTTP 请求路径封死了，用户无法直接调用外部库，才不得不等"大神"写一个适配层。每个 scrapy-xxx 插件的存在，都是对框架自身局限性的一次补丁。

2. 插件带来的隐性成本远大于便利：

   • 版本冲突：scrapy-playwright 可能和你的 scrapy 版本不兼容

   • 作者弃坑：很多 scrapy 插件多年无人维护

   • 更新滞后：Playwright 出新版，你得等插件作者跟进适配，可能等数月

   • 额外学习：每个插件有自己的配置项和坑，等于学了 scrapy 还得再学一套插件 API

3. Funboost 的做法本质上更优越：

   • Playwright 出新版？直接 pip install --upgrade playwright，你的爬虫函数立即可用最新特性

   • 想用 httpx？直接 import，不需要等 "scrapy-httpx" 这种东西被人开发出来

   • 想用 curl_cffi 绕 TLS 指纹？直接 import，scrapy 生态里可能永远都不会有人写这个插件

结论：Funboost 不需要插件不是因为"生态不成熟"，而是因为设计理念在更高维度——整个 Python 生态（PyPI 上的几十万个包）天然就是 funboost 的"插件库"，无需任何适配层。这比 scrapy 的封闭插件体系强大无数倍。

---

8.4 Funboost 爬虫辅助工具：funspider 与 boost_spider

虽然 Funboost 本身不限制你用什么 HTTP 库和解析库，但为了进一步提升爬虫开发效率，提供了两个爬虫增强方案：

• funboost = 分布式函数调度引擎：分布式消费、任务去重、失败重试、精准QPS控频、分布式全局控频、ACK可靠消费、断点续爬、死信队列、消息过期、函数超时、定时任务、延时任务、链式派发、任务裂变、多进程叠加并发、RPC、FaaS微服务、可视化管理、监控告警、远程部署、熔断、周期额度、热加载、跨系统联动……

• boost_spider = boost_spider = RequestClient + SpiderResponse + DatasetSink

• funspider = SimpleSpiderClient/AsyncSpiderClient + SpiderResponse + SpiderItem

8.4.1 funspider（内置于 funboost）

funspider 是 funboost 自带的爬虫辅助扩展，位于 funboost/contrib/funspider/。适合偏好 ORM 模型驱动 + 异步协程 的开发者。

核心组件：

组件	说明
SimpleSpiderClient	同步爬虫客户端（基于 httpx），内置自动重试、随机 UA、代理轮换
AsyncSpiderClient	异步爬虫客户端（基于 httpx），适配 funboost 的 ASYNC 并发模式
SpiderResponse	响应对象，自带 .xpath() / .css() / .re() / .re_first() / .resp_dict
SpiderItem	基于 SQLModel 的 ORM 模型，支持 upsert() / aio_upsert() / mongo_upsert() / bulk_upsert()

完整代码示例：

from funboost import boost, BoosterParams, BrokerEnum
from funboost.contrib.funspider import SimpleSpiderClient, SpiderItem, Field, create_engine

# 1. 定义 ORM 数据模型
class NewsItem(SpiderItem, table=True):
    __tablename__ = "news"
    __engine__ = create_engine("mysql+pymysql://root:123@localhost/spider_db")
    __default_upsert_unique_fields__ = ["news_id"]

    id: int = Field(default=None, primary_key=True)
    news_id: int = Field(unique=True)
    title: str = Field(max_length=200)
    url: str = Field(max_length=500)

# 2. 创建客户端（自动换 UA + 代理轮换 + 失败重试）
def abuyun_proxy():
    return "http://user:pass@proxy.abuyun.com:9020"

client = SimpleSpiderClient(proxy_getter_list=[abuyun_proxy], retry_times=3)

# 3. 编写爬虫函数
@boost(BoosterParams(queue_name='news_list', broker_kind=BrokerEnum.REDIS_ACK_ABLE, qps=2))
def crawl_list(page: int):
    resp = client.get(f"https://news.example.com/list?page={page}")
    for link in resp.css("a.title::attr(href)").getall():
        crawl_detail.push(url=link)

@boost(BoosterParams(queue_name='news_detail', broker_kind=BrokerEnum.REDIS_ACK_ABLE, qps=5, max_retry_times=5))
def crawl_detail(url: str):
    resp = client.get(url)
    news_id = int(resp.re_first(r'/news/(\d+)'))
    title = resp.xpath("//h1/text()").get()
    NewsItem(news_id=news_id, title=title, url=url).upsert()  # 一行入库

# 4. 启动
if __name__ == "__main__":
    crawl_list.consume()
    crawl_detail.consume()
    for page in range(1, 100):
        crawl_list.push(page=page)

安装依赖：

pip install funboost httpx parsel sqlmodel

8.4.2 boost_spider（独立三方包）

pip install boost_spider

boost_spider 是基于 funboost 的爬虫增强包，提供三大核心类：

组件	说明
RequestClient	爬虫请求客户端（API 兼容 requests），支持 多代理商容灾轮换（阿布云挂了自动切快代理）、自动重试、随机 UA、Cookie 会话保持
SpiderResponse	响应自带 .xpath() / .css() / .re_search()
DatasetSink	一行代码将字典保存到 MySQL/PostgreSQL/SQLite/MongoDB，自动建表

funspider vs boost_spider 选型：

维度	funspider	boost_spider
设计理念	ORM 辅助、强类型流	自由至上、极简字典流
数据模型	SQLModel 类，IDE 自动补全	纯字典，DatasetSink 一行入库
HTTP 客户端	httpx (同步+异步双引擎)	API 兼容 requests，多代理商容灾
适合场景	大型团队项目、需要 ORM 管理表结构	快速开发、极致简洁、完全掌控

两者都是锦上添花的辅助工具，不是必须的。即使不用它们，直接 import requests + import parsel 写在函数里也完全 OK。

---

8.5 Funboost vs Scrapy：关键场景实战对比

8.5.1 反爬换代理 IP

Scrapy 方式（需要精通框架生命周期）：

# middlewares.py
class ProxyMiddleware:
    def process_request(self, request, spider):
        request.meta['proxy'] = random.choice(proxy_list)

# settings.py
DOWNLOADER_MIDDLEWARES = {'myproject.middlewares.ProxyMiddleware': 543}

Funboost 方式（纯 Python 直觉）：

@boost(BoosterParams(queue_name='my_crawl', max_retry_times=3))
def crawl(url):
    proxy = random.choice(proxy_list)
    ua = random.choice(ua_list)
    resp = requests.get(url, proxies={'http': proxy}, headers={'User-Agent': ua})
    # ...

差异本质：Scrapy 需要学习中间件机制才能实现，Funboost 只需要 Python 基础语法。

8.5.2 短时效 Token（5秒内过期）

场景：先请求 API 获取 Token，必须在 5 秒内用这个 Token 请求另一个 API。

Scrapy：两次请求被拆成两个 yield Request + callback，经过调度器排队，无法保证时序，Token 极易过期。

Funboost：在同一个函数内线性执行，天然保证时序：

@boost(BoosterParams(queue_name='token_crawl', qps=2))
def crawl_with_token(data_id):
    token_resp = requests.post('/api/token', json={'id': data_id})
    token = token_resp.json()['token']  # 获取 Token
    # 立即使用，不到1毫秒延迟
    data_resp = requests.get(f'/api/data?token={token}&id={data_id}')
    save(data_resp.json())

8.5.3 浏览器多轮交互

场景：输入文字 → 点击按钮 → 等待 → 判断内容 → 再点击 → 提取数据。

Scrapy：几乎不可能实现。Selenium 会阻塞 Twisted 事件循环，把异步框架退化成单线程。

Funboost：在函数里自然编写，框架自动并发 N 个浏览器实例：

@boost(BoosterParams(queue_name='browser_task', concurrent_num=5, function_timeout=60))
def interact_and_crawl(keyword):
    from playwright.sync_api import sync_playwright
    with sync_playwright() as p:
        page = p.chromium.launch().new_page()
        page.goto('https://search.example.com')
        page.fill('#input', keyword)
        page.click('#submit')
        page.wait_for_selector('.results')
        if page.query_selector('.captcha'):
            page.click('.verify-btn')
            page.wait_for_timeout(3000)
        data = page.query_selector_all('.item')
        # ... 提取并存储

8.5.4 动态实时添加任务（微服务场景）

场景：Java 后端需要实时触发 Python 爬虫去抓一条数据。

Scrapy：封闭黑盒，外部系统无法直接注入任务，需要魔改代码或数据库中转。

Funboost：天然支持。Java 只需往 Redis/RabbitMQ 对应的 queue_name 发一条 JSON（内容就是函数入参）：

{"url": "https://example.com/product/123"}

Funboost 消费者立即执行。也可用 funboost.faas 一键生成 HTTP 接口，让外部系统通过 REST API 触发爬虫。

---

8.6 函数级重试 vs URL级重试

这是 Funboost 在爬虫可靠性上的核心优势之一。

Scrapy 的 URL 级重试：只在 HTTP 请求失败（超时、5xx）时重试。如果服务器返回 HTTP 200 但内容是反爬验证码或空数据，Scrapy 认为"成功了"，数据直接丢失。

Funboost 的函数级重试：函数内任何位置抛出异常都会触发重试。这意味着：

@boost(BoosterParams(queue_name='reliable_crawl', max_retry_times=5))
def crawl(url):
    resp = requests.get(url)
    data = resp.json()
    if data.get('code') == 'captcha_required':  # HTTP 200 但业务反爬
        raise Exception("触发验证码")  # 自动重试，换 IP 再来
    title = data['result']['title']  # KeyError 也会触发重试
    save(title)

即使是 解析代码的 bug 导致的 KeyError、IndexError，也会被框架捕获并自动重试（换代理/换UA后可能成功）。Scrapy 在这种场景下直接丢数据。

---

8.7 讨Scrapy檄文：Funboost兴，Scrapy亡，天下爬虫，当顺天命！

盖闻：天地有正道，万物循至理。编程之道，贵在通达；调度之术，胜在自由。

昔者Scrapy挟Twisted之余烈，立URL调度之宗庙，号令爬虫诸侯十有余年。然时移世易，其法渐陈：回调为锁，中间件为牢，Pipeline为械。天下开发者苦Scrapy久矣——代码支离破碎，调试如堕迷雾，插件卡脖，旧码难容，分布式如空中楼阁，生产环境如履薄冰！

今有 Funboost 横空出世！以函数为剑，以@boost为令，化繁为简，破旧立新。一行装饰器，赋万物以分布式之力；一个函数体，纳天下PyPI为我所用。此诚天道更替、不可逆之大势也！

吾观Scrapy之根基，十败已定；Funboost十胜在手，当为天下主！

---

一曰：道失对道胜！ Scrapy以URL为心，以Request为锁，调度一次请求便自诩功成。殊不知URL不过函数之一行耳！框架替用户发请求，用户反被框架所缚，此谓 "画地为牢困蛟龙"！ Funboost以函数为道，万法归宗。调度一个函数，函数之内可行万里路、做万般事——发请求、操浏览器、调模型、写数据库，无所不能，此谓 "道法自然御星河"！

二曰：繁失对简胜！ Scrapy开工先筑城：scrapy.cfg、items.py、pipelines.py、middlewares.py、settings.py、spiders/... 七八文件列阵排兵，仪式感重于泰山，初学者望而却步，此谓 "千乘列阵烹一鸡"！ Funboost一@boost点睛，单文件搞定一切。验证函数能跑，加上装饰器即刻起飞，此谓 "一字点化龙乘风"！

三曰：弱失对强胜！ Scrapy困于单进程，Twisted虽号异步，遇Selenium则英雄气短，异步退化同步，多核CPU空置叹息，此谓 "独木难撑万钧鼎"！ Funboost多线程/协程叠加多进程，再叠多机器，四重并发层层递进。百台机器万核齐鸣，势如破竹，此谓 "千军万马踏九州"！

四曰：粗失对精胜！ Scrapy仅有DOWNLOAD_DELAY聊以控频，受响应耗时波动，QPS漂移不定，控频如盲人射箭，此谓 "刻舟求剑失方寸"！ Funboost基于精密计时算法控频，qps=5即每秒恰好5次，且能自动补偿网络抖动；分布式场景下多机器自动均分配额，全局精准联控，此谓 "庖丁解牛游刃余"！

五曰：乱失对明胜！ Scrapy回调嵌套层叠，yield Request(callback=parse_2, meta={...}) 再 yield Request(callback=parse_3, meta={...})，逻辑碎片化如散沙，meta字典传参无类型、无补全、拼错无警告，此谓 "盲骑瞎马临深池"！ Funboost函数平铺直叙，从上到下一气呵成。局部变量有类型，IDE自动补全，重构一键同步，此谓 "长河落日照秋明"！

六曰：虚失对固胜！ Scrapy-redis以blpop取消息，取出即离队。进程崩溃、服务器断电，内存中数十万任务灰飞烟灭，断点续爬形同虚设，此谓 "累卵高筑一风倾"！ Funboost消费确认(ACK)机制——消息处理成功才从队列移除。崩溃重启，未完成任务自动回队重投，万无一失，此谓 "磐石不动海潮平"！

七曰：枯失对荣胜！ Scrapy需scrapy-redis才能分布式，需scrapy-selenium才能操浏览器，需scrapy-playwright才能用Playwright。无插件则寸步难行，等插件如望穿秋水，此谓 "缘木求鱼空嗟叹"！ Funboost无需任何插件！天下PyPI十数万包皆为己用——import requests、import playwright、import pymongo，直接调用，天然兼容，此谓 "万物为我所驱遣"！

八曰：漏失对全胜！ Scrapy重试仅针对HTTP请求失败（超时、5xx）。若服务器返回HTTP 200却是验证码或空数据，Scrapy认为"成功"，数据永久丢失！此谓 "金玉其外败絮中"！ Funboost函数级重试——函数内任意位置抛异常即触发重试。HTTP 200但内容异常？解析报错？一律自动换IP重来，数据滴水不漏，此谓 "天网恢恢疏不漏"！

九曰：闭失对通胜！ Scrapy是封闭孤岛。Java后端想实时触发爬虫？运营想按需抓一条数据？只能魔改代码或数据库中转，延迟极高，耦合极重，此谓 "闭关锁国难通商"！ Funboost天然是微服务——外部系统直接往消息队列发一条JSON即可触发爬取。Java、Go、PHP随时调用，funboost.faas一键生成HTTP接口，此谓 "四海通衢万邦来"！

十曰：拙失对巧胜！ Scrapy遇短时效Token（5秒过期）、浏览器多轮交互、动态IP加密签名等奇巧需求，则左支右绌、力不从心。回调机制无法保证时序，复杂流程在框架中寸步难行，此谓 "削足适履步维艰"！ Funboost函数之内，时序确定、状态自然、连续操作一步到位。获取Token后立即使用、多轮点击等待判断一气呵成，此谓 "千变万化指掌间"！

---

有此十胜，Funboost伐Scrapy，如沸汤泼雪，如皓月蔽萤，何愁不克？！

Scrapy老将，若审时度势弃旧图新，尚可涅槃重生；若抱残守缺负隅顽抗，必为时代洪流碾为齑粉！

今Funboost大纛已立，携函数调度之神锋，集分布式并发之铁骑： 东纳Requests之勇，西引HTTPX之锐；南征Selenium之坚，北驭Playwright之奇！ 以Redis、RabbitMQ、Kafka近五十种消息队列为粮道，合多进程、多线程、协程三军为一体！

天下英雄，当明辨大势，顺道而行！ 弃yield Request之桎梏，拥@boost之光辉！ 自此开发者无需叩拜Spider宗庙，无需忍受回调地狱之煎熬，无需跪求插件大神之垂怜！ 此乃爬虫之文艺复兴，调度之工业革命！

剑锋所指，框架枷锁灰飞烟灭！函数光辉，普照四海八荒！

与诸君共勉，同开万物皆可Boost之盛世！

FunBoost大都督，昭告天下！

8.8 funspider介绍

boost_spider是一个单独的项目依赖funboost驱动，funspider是funboost下的一个贡献模块。

funspider 不是爬虫框架,也绝对不是为funboost自身量身定制的紧耦合的插件，它是基于 funboost 分布式函数调度引擎的爬虫辅助层，只是增加请求 和 响应解析 和保存数据库三个类。funboost 有多强大，funspider 就有多强大。百分之百利用funboost的所有功能,例如50种消息队列 + 5种并发方式 + 30种任务控制功能 + funweb可视化管理 完全可以利用。

用法直接看funboost源码中的文件的demo例子 funboost\contrib\funspider\funspider_demos\funspider_demo1.py

8.9 funboost + funspider + boost_spider + your_utils 混合使用

funspider 和 boost_spider 的三个类都是很普通的工具类，绝对没有和funboost有任何紧耦合的关系，所以可以和funboost自由搭配，用户可以自由使用 funboost + funspider(部分类) + boost_spider(部分类) + your_utils(用户自己团队历史沉淀打磨的工具类) 来写爬虫。

• funboost = 分布式函数调度引擎：分布式消费、任务去重、失败重试、精准QPS控频、分布式全局控频、ACK可靠消费、断点续爬、死信队列、消息过期、函数超时、定时任务、延时任务、链式派发、任务裂变、多进程叠加并发、RPC、FaaS微服务、可视化管理、监控告警、远程部署、熔断、周期额度、热加载、跨系统联动……

• boost_spider = boost_spider = RequestClient + SpiderResponse + DatasetSink

• funspider = SimpleSpiderClient/AsyncSpiderClient + SpiderResponse + SpiderItem

## 8.9 funboost + funspider + boost_spider + your_utils 混合使用

funspider 和 boost_spider 的三个类都是很普通的工具类，绝对没有和funboost有任何紧耦合的关系，所以可以和funboost自由搭配，用户可以自由使用 funboost + funspider(部分类) + boost_spider(部分类) + your_utils(用户自己团队历史沉淀打磨的工具类) 来写爬虫。

```python
"""
funboost + boost_spider(RequestClient) + funspider(SpiderItem ORM) + your_utils   混合示例

四个组件各管一件事：
funboost:      分布式调度、去重、重试、QPS控制        → @boost(BoosterParams(...))
boost_spider:  HTTP请求(代理/UA轮换)、HTML解析       → RequestClient (返回 SpiderResponse)
funspider:     ORM数据模型、类型校验、upsert入库      → SpiderItem / Field / create_engine
your_utils:    用户自己的业务工具(加密/签名等)         → decrypt_sign / encrypt_token
"""
from typing import Optional

# 分别从各自包导入，体现"四组件混合"
from funboost import boost, BoosterParams, BrokerEnum, BoostersManager, enable_ctrl_c_quit_on_windows, ApsJobAdder
from boost_spider import RequestClient, SpiderResponse
from funboost.contrib.funspider import SpiderItem, Field, create_engine
from your_utils import decrypt_sign, encrypt_token  # 用户自己的工具函数

CRAWLER_GROUP = "news_crawler_group"  # 消费分组名，所有函数必须写同一个，用于一键启动

# ==================== 第一步：定义 ORM 数据模型 (funspider) ====================

ENGINE = create_engine("sqlite:///news.db")


class NewsItem(SpiderItem, table=True):
    """
    继承 SpiderItem，table=True 表示这是一个真实的数据库表。
    字段类型直接写 Python 类型注解，SQLModel 自动映射为数据库列。
    """
    __tablename__ = "news"
    __engine__ = ENGINE
    __default_upsert_unique_fields__ = ["url"]  # upsert 去重字段：url 相同则更新，不同则插入

    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    url: str
    publish_time: str
    content: str


NewsItem.create_table()  # 自动建表，已存在则跳过


# ==================== 第二步：创建 HTTP 请求客户端 (boost_spider) ====================

# RequestClient 全局复用，内部自动维护 Session/cookie 状态
client = RequestClient(
    request_retry_times=3,              # 请求失败自动重试 3 次
    is_change_ua_every_request=True,    # 每次请求自动随机切换 User-Agent
)


# ==================== 第三步：写爬虫函数 + 加 funboost 调度 ====================

@boost(BoosterParams(
    queue_name='news_list',
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,   # Redis 队列 + ACK 确认（防丢任务）
    qps=3,                                    # 每秒最多消费 3 个任务
    do_task_filtering=True,                   # 开启任务去重（相同参数不重复执行）
    task_filtering_expire_seconds=3600 * 24,  # 去重有效期 24 小时
    max_retry_times=5,                        # 函数内抛异常自动重试最多 5 次
    booster_group=CRAWLER_GROUP,              # ★ 消费分组，必须写
))
def crawl_list(category_id: int):
    """爬取某个分类的列表页，解析出所有详情页链接，推送到详情页任务"""

    # --- your_utils: 调用用户自己的加密工具 ---
    sign = decrypt_sign(category_id)

    # --- boost_spider: 发请求，返回的 resp 直接就是 SpiderResponse ---
    resp = client.request('get', 'https://example.com/api/list', params={'cat': category_id, 'sign': sign})

    # --- boost_spider: SpiderResponse 直接支持 xpath 解析 ---
    # xpath 返回 SelectorList，用 .getall() 取出所有匹配的字符串
    for link in resp.xpath('//a[@class="news-link"]/@href').getall():
        news_id = link.split('/')[-1]
        # funboost: .push() 推种子到详情页队列，自动去重
        crawl_detail.push(news_id)


@boost(BoosterParams(
    queue_name='news_detail',
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    qps=2,
    do_task_filtering=True,
    task_filtering_expire_seconds=3600 * 24 * 7,  # 详情页去重有效期 7 天
    max_retry_times=5,
    booster_group=CRAWLER_GROUP,                   # ★ 同一个消费分组
))
def crawl_detail(news_id: str):
    """爬取单个详情页，解析内容，用 upsert 入库（url 相同则更新，不重复插入）"""

    # --- your_utils: 调用用户自己的 token 工具 ---
    token = encrypt_token(news_id)

    # --- boost_spider: 发请求 ---
    resp = client.request('get', f'https://example.com/news/{news_id}', headers={'x-token': token})

    # --- funspider: ORM 实例化 + upsert 入库 ---
    # upsert 逻辑：__default_upsert_unique_fields__ = ["url"]
    #   → 数据库中已有这个 url → 更新已有记录
    #   → 数据库中没有这个 url → 插入新记录
    NewsItem(
        title=resp.xpath('//h1/text()').extract_first(),
        url=f'https://example.com/news/{news_id}',
        publish_time=resp.css('.pub-time::text').extract_first(),
        content=resp.xpath('//div[@class="content"]//text()').extract_first(),
    ).upsert()


# ==================== 第四步：推种子 + apsjobadder 定时推送 + 启动消费 ====================

@boost(BoosterParams(
    queue_name='news_seed_scheduler',
    broker_kind=BrokerEnum.REDIS,
    booster_group=CRAWLER_GROUP,
))
def push_daily_seeds():
    """批量推送 1~10 分类种子；既可立即调用，也可作为定时任务目标"""
    for cat_id in range(1, 11):
        crawl_list.push(cat_id)


if __name__ == '__main__':
    # ★ 启动消费组：一次性启动 crawl_list、crawl_detail、push_daily_seeds 三个消费函数
    BoostersManager.consume_group(CRAWLER_GROUP)

    # ★ 启动后先立即推一次种子，便于首次运行马上看到效果
    push_daily_seeds.push()

    # ★ 使用 apsjobadder 每天 0 点定时 push 一个 booster 任务，再由该任务批量推送 1~10 分类种子
    # 这个相当于导火索 引线，每天0点触发一个 push_daily_seeds定时任务，引爆 列表页 详情页 链式派发，大规模执行，像一个导火索。
    # 在爬虫种子裂变上，funboost是祖宗，scrapy feapder是小弟。
    ApsJobAdder(push_daily_seeds, job_store_kind='redis').add_push_job(
        trigger='cron',
        hour=0,
        minute=0,
        id='push_all_news_categories_daily',
        replace_existing=True,
    )

    # 可选：加上后方便 Windows 系统按 Ctrl+C 退出
    enable_ctrl_c_quit_on_windows()

---

8.10 集中对比：Funboost 30项 vs Scrapy

维度	Funboost	Scrapy
调度核心	调度函数（内部完全自由）	调度 Request 对象（框架控制）
编程范式	平铺直叙线性代码	yield Request + callback 回调链
状态管理	函数内局部变量	response.meta 字典传递
框架侵入性	仅 @boost 装饰器	必须继承 scrapy.Spider
项目结构	无要求，单文件即可	强制 7-8 个文件
学习曲线	几分钟上手	需学生命周期/中间件/Pipeline 等整套体系
HTTP库选择	任意（requests/httpx/aiohttp/playwright...）	强制 Twisted 下载器
反爬实现	普通函数封装，0门槛	需编写 Downloader Middleware
单元测试	直接调用函数测试	与框架耦合，难以独立测试
IDE 补全	函数参数完整补全	response.meta 无法补全
调试	标准 pdb/IDE 调试	回调链+异步，调试困难
并发模型	多进程×多线程/协程×多机器 四重叠加	单进程受限，多核利用差
速率控制	精准 QPS 控制	DOWNLOAD_DELAY 不精确
分布式控频	跨机器全局 QPS 限制	不支持
复杂流程	函数内多轮交互自然编写	回调处理笨拙，阻塞异步模型
短时效 Token	函数内连续请求保证时效	调度器排队导致 Token 过期
任务去重	基于函数入参智能去重	URL 指纹去重，噪音参数干扰
去重有效期	支持	不支持
错误重试	函数级（HTTP 200 内容错误也重试）	URL 级（仅网络错误重试）
断点续爬	消费确认(ACK)，崩溃不丢消息	scrapy-redis blpop 丢消息
消息队列	近 50 种	主要 scrapy-redis
定时任务	原生支持 Cron	需外部工具
数据持久化	函数内直接调用任何库	必须 Item Pipeline
插件生态	无需插件，PyPI 所有包直接用	依赖专用插件（scrapy-xxx）
跨语言交互	JSON 消息队列，Java/Go 直接发任务	不支持
Web 管理	funboost web manager 全功能	无官方工具
远程部署	内置 fabric_deploy 一键部署	无此功能
RPC 模式	支持同步等待爬取结果	不支持
微服务/FaaS	原生支持外部注入任务	封闭系统，难以对接
代码复用	utils/ 现有函数零成本复用	必须改成 Middleware 形状

---

8.11 Funboost 爬虫最佳实践

8.11.1 项目结构推荐

Funboost 不强制任何目录结构。推荐的做法是：一个网站一个 py 文件，列表页+详情页的函数都写在同一个文件中，在文件底部的 if __name__ 里启动消费。一个项目可以爬几十个不同网站。

请求和存储可以选择：

• 自己封装：用你 utils/ 下的 my_request 函数（完全自由，适合已有积累的团队）

• 用 funspider：内置 SimpleSpiderClient / AsyncSpiderClient（自带代理轮换+随机UA+自动重试+Cookie会话保持+响应内置xpath/css）+ SpiderItem（一行 ORM upsert 入库）

• 用 boost_spider：RequestClient（多代理商容灾+自动重试+Cookie会话保持）+ DatasetSink（一行代码自动建表存库）

三种方式任选，funboost 都不限制。

my_spider_project/
├── utils/
│   ├── my_request.py      # 方式1：你自己封装的通用请求函数
│   └── db.py              # 你自己的数据库存储工具
├── site_autohome.py       # 汽车之家爬虫（列表+详情，同一文件）
├── site_zhihu.py          # 知乎爬虫（列表+详情，同一文件）
├── site_douban.py         # 豆瓣爬虫（列表+详情，同一文件）
└── run_all.py             # 可选：一键启动所有网站的消费

8.11.2 单文件完整示例

# site_autohome.py —— 一个文件搞定汽车之家新闻爬虫
from funboost import boost, BoosterParams, BrokerEnum
from utils.my_request import my_request

@boost(BoosterParams(queue_name='autohome_list', broker_kind=BrokerEnum.REDIS_ACK_ABLE, qps=2))
def crawl_list(news_type: str, page: int):
    resp = my_request('get', f'https://www.autohome.com.cn/{news_type}/{page}/')
    for url in parse_list_urls(resp):
        crawl_detail.push(url=url)
    if page == 1:
        for p in range(2, get_last_page(resp) + 1):
            crawl_list.push(news_type=news_type, page=p)

@boost(BoosterParams(queue_name='autohome_detail', broker_kind=BrokerEnum.REDIS_ACK_ABLE, qps=5, max_retry_times=3))
def crawl_detail(url: str):
    resp = my_request('get', url)
    data = parse_detail(resp)
    save_to_db(data)

if __name__ == "__main__":
    crawl_list.consume()
    crawl_detail.consume()
    # 发布种子任务
    crawl_list.push(news_type='news', page=1)
    crawl_list.push(news_type='drive', page=1)

运行就一条命令：python site_autohome.py。不需要额外的启动脚本。

8.11.3 多进程 + 多线程极限性能

# 8核机器，每进程50线程 = 400并发
crawl_detail.multi_process_consume(8)

8.11.4 精准 QPS 控频（保护目标网站）

@boost(BoosterParams(
    queue_name='polite_crawl',
    qps=2,  # 每秒最多2次请求
    is_using_distributed_frequency_control=True,  # 多台机器合计不超过 2 QPS
))
def crawl(url): ...

8.11.5 funweb 可视化管理爬虫（开箱即用）

funboost 自带 Web 管理界面 funweb，一条命令启动（pip install funboost[flask] && python3 -m funboost.funweb.app），浏览器打开 127.0.0.1:27018 即可使用。也可在代码中调用 start_funboost_web_manager() 随爬虫一起启动。

因为 Funboost 一个函数执行 = 爬一次页面，所以 funweb 的消费指标天然就是爬虫指标：

爬虫监控核心能力：

• 消费速率曲线：实时+历史曲线展示每个爬虫的爬取速度（按秒/分钟/小时自动切粒度），含近10秒完成数、失败数、平均耗时、剩余积压等 7 项指标

• 暂停/恢复消费：发现被封 IP？一键暂停，换完 IP 再恢复

• 消费结果查询 + 失败重投：查看每个 URL 的爬取结果（成功/失败/异常堆栈/耗时），失败任务一键重新投递

• 告警推送：积压超标、QPS 骤降、消费者掉线、失败率飙升、耗时过高——5 种告警自动推送到企业微信/钉钉/飞书

• 脚本部署：内置 Supervisor + Jenkins + ELK 三合一——一键部署爬虫、自动保活、Git Pull 一键发版、10GB+ 日志毫秒级检索（彩色高亮）

• 定时任务：页面配置 Cron 定时爬取，无需写系统 crontab

对比 Scrapy：Scrapy 没有官方 Web 管理界面。想要类似的监控+部署+告警能力，需要自己搭建 Grafana + Prometheus + Supervisor + Jenkins + ELK 一整套体系。funweb 一条命令全部搞定。

---

8.12 总结

选 Funboost 还是 Scrapy？
你想写函数就能爬虫，享受自由	→ Funboost
你想复用自己 utils/ 文件夹的宝贵资产	→ Funboost
你需要精准 QPS 控频 + 分布式全局限流	→ Funboost
你需要处理短时效 Token / 浏览器多轮交互	→ Funboost
你需要其他部门通过 API 实时触发爬取	→ Funboost
你需要崩溃/断电后消息万无一失	→ Funboost
你需要多进程 × 多线程 × 多机器极限性能	→ Funboost

Funboost 的核心优势不是"对爬虫做了什么专门优化"，而是"根本不限制你做什么"。 它用更高维度的"通用函数调度"解决了爬虫问题，给予开发者完全的自由，同时自动赋予 30+ 种企业级控制能力。这是对任何"框架束缚型"爬虫工具的降维打击。