RabbitMQ不能直接做去重队列,因其不支持消息内容级去重,message_id仅单次有效,无法跨生产者/连接/重启识别重复;去重必须前置(如Redis SET/BloomFilter)或后置(消费端查库),RabbitMQ仅负责可靠分发与负载均衡。
为什么不能直接用 RabbitMQ 做去重队列
rabbitmq 本身不提供消息内容级去重能力,message_id 需手动设置且仅在单次发布中生效,无法跨生产者、跨连接或跨重启判断重复。如果你把 url 当作消息体直接发进队列,又没做外部判重,爬虫集群会反复抓取同一 url——这不是 rabbitmq 的缺陷,而是它定位本就不是去重中间件。
实际做法是:RabbitMQ 只负责可靠分发和负载均衡,去重必须前置(如用 Redis 的 SET 或 BloomFilter),或后置(消费端查库再决定是否处理)。
- 去重逻辑必须独立于 RabbitMQ,推荐在生产者端接入
Redis:发消息前先redis.sismember("seen_urls", url),命中则跳过 - 若 URL 量极大(亿级),用
pybloom_live+Redis做两层布隆过滤,降低误判率 - 不要依赖 RabbitMQ 的
x-message-ttl或死信机制来“过滤重复”,它们不解决语义重复问题
如何设计 URL 调度的 Exchange 和 Queue 结构
用 RabbitMQ 做分布式调度,核心是让多个爬虫 worker 公平竞争 URL,同时支持动态扩缩容。推荐使用 direct 或 topic Exchange,配合 durable 队列和手动 ack,避免 worker 挂掉导致 URL 丢失。
典型结构:exchange="url_dispatch"(direct 类型),绑定多个队列如 "spider_a_queue"、"spider_b_queue",路由键(routing_key)可按域名或优先级划分(如 "news.sina.com.cn"、"priority.high")。
- 所有生产者向
url_dispatch发送消息,routing_key决定投递目标队列 - 每个 worker 启动时声明专属队列并绑定对应
routing_key,实现逻辑隔离 - 队列必须设
durable=True,消息发送时设delivery_mode=2,保证宕机不丢 URL - 消费端务必用
channel.basic_qos(prefetch_count=1)控制并发,防止某个 worker 积压太多未确认 URL
消费端如何安全处理 URL 并防止重复抓取
即使生产端做了去重,网络重试、worker 重启、消息重入仍可能导致同一条 URL 被多次消费。安全做法是:消费即锁定 + 处理完再落库 + 异常时拒绝而非丢弃。
Python 3.14.3
微软官方的 Python 扩展,是 VS Code 安装量最高的扩展(209M+)。集成 IntelliSense(通过 Pylance)、调试(通过 Python Debugger)、代码检查、格式化、重构和单元测试等功能。支持 Jupyter Notebook、虚拟环境管理和多 Python 版本切换。
关键动作顺序不能错:接收到消息 → 从 body 解析 url → 查 Redis 或数据库确认未抓取 → 发起 HTTP 请求 → 解析结果 → 存储 → 标记该 url 为已抓取 → 才调用 channel.basic_ack()。
- 如果中间任何一步失败(如解析异常、存储超时),必须调用
channel.basic_nack(requeue=False),让消息进死信队列,而不是requeue=True导致无限循环 - 死信队列要单独监听,人工介入或加延迟重试(通过
x-dead-letter-exchange+delayed message plugin) - 不要在消费逻辑里直接写文件或调用慢接口,URL 处理路径必须轻量;重任务拆到后续 pipeline(如用另一个 RabbitMQ 队列传给解析服务)
Python 客户端用 pika 还是 aio-pika?
同步爬虫(如基于 requests + lxml)用 pika 更稳妥;异步爬虫(aiohttp + asyncio)必须用 aio-pika,否则会阻塞事件循环。
pika 默认是 blocking connection,适合单 worker 单线程场景;但要注意它不支持真正的异步 I/O,高并发下需配合多进程(multiprocessing)而非多线程(GIL 限制)。
- 用
pika时,每个 worker 进程应独占一个BlockingConnection,别在线程间共享 channel - 用
aio-pika时,注意channel.close()是协程,需await;消息体默认是 bytes,记得用json.loads(body.decode()) - 无论哪种,都必须捕获
AMQPConnectionError和ChannelClosedByBroker,实现自动重连逻辑,否则网络抖动会导致整个 worker 卡死
RabbitMQ 在爬虫调度里真正难的不是怎么连,而是怎么跟去重、重试、监控、限速这些环节咬合严实。最容易被忽略的是消息生命周期管理:从生产者写入 Redis 判重,到 RabbitMQ 持久化,再到消费者手动 ack 和死信兜底,任意一环松动,就会漏 URL 或重复抓。