Python 爬虫系统如何利用 RabbitMQ 作为分布式分布式 URL 调度中心和去重队列？-Python教程-PHP中文网

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-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 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 卡死