直接用pika发送任务会丢消息,是因为默认未启用发布确认、队列未持久化、消息未设delivery_mode=2;漏掉任一机制,RabbitMQ重启或消费者异常时消息即丢失。
为什么直接用 pika 发送任务会丢消息?
默认情况下,pika 的 channel.basic_publish() 不保证消息持久化,rabbitmq 重启或消费者异常退出时,未消费的消息直接丢失。这不是代码写错了,而是 rabbitmq 的默认行为——消息只存在内存中。
必须显式开启三个机制才能真正“不丢”:
-
connection开启confirm=True(启用发布确认) -
channel.queue_declare()中设置durable=True(队列持久化) -
channel.basic_publish()中设置properties=pika.BasicProperties(delivery_mode=2)(消息持久化)
漏掉任意一项,都可能在压力测试或服务重启后发现任务“凭空消失”。尤其注意 delivery_mode=2 是整数,不是字符串 "2",否则静默失效。
爬虫 worker 怎么避免重复消费同一 URL?
RabbitMQ 本身不负责去重,URL 去重必须由业务层控制。常见错误是把去重逻辑放在 worker 启动时加载一次全局 set,但多进程下每个子进程都有独立内存空间,互相不可见。
更可靠的做法是:在消费前,用 Redis 的 SETNX 或 PFADD 做原子判重:
立即学习“Python免费学习笔记(深入)”;
import redis
r = redis.Redis()
url = "https://example.com/page/123"
if r.setnx(f"seen:{hash(url) % 1000}", url): # 分桶避免单 key 过大
parse_page(url)
else:
print(f"skip duplicated {url}")
注意不要用 r.sismember("urls_seen", url) + r.sadd() 两步,中间有竞态;也别依赖 RabbitMQ 的 ack 时机做去重,因为网络延迟可能导致重复投递(at-least-once 语义)。
如何让 worker 自动重连并恢复未完成任务?
网络抖动或 RabbitMQ 临时不可用时,worker 不能直接 crash,否则正在处理但未 ack 的消息会被 RabbitMQ 重新入队——如果没设 requeue=False,可能无限循环重试同一失败任务。
正确做法是:
- 捕获
pika.exceptions.AMQPConnectionError和pika.exceptions.ChannelClosedByBroker,触发重连逻辑 - 消费回调函数里用
try/except包裹解析逻辑,出错时调用channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False) - 把失败的 URL 写入单独的
failed_urls队列,供人工检查或定时重试
别依赖 auto_ack=True 省事——它等于放弃可靠性,消息一发就删,worker 处理到一半挂掉就再也找不回来了。
用 celery 还是裸写 pika?
如果你只需要分发 URL、返回解析结果、无定时调度、无任务优先级、无复杂依赖链,裸写 pika 更轻量、可控性更强。Celery 默认用 pickle 序列化,爬虫常要传 requests.Session 或自定义类,容易序列化失败;而且 Celery 的 broker heartbeat 检测有时比实际网络延迟还激进,导致假断连。
但如果需要动态扩缩容 worker、查看任务历史、集成 Flower 监控、或未来加定时抓取(beat),Celery 的抽象就值得引入。此时务必改用 json 序列化:CELERY_TASK_SERIALIZER = 'json',并确保所有 task 参数都是基础类型(str/int/dict)。
最易被忽略的是 connection 复用:每个 worker 进程应只创建一个 pika.BlockingConnection,反复 reuse,而不是每次消费都新建连接——RabbitMQ 对连接数有限制,且 TCP 握手开销远大于消息本身。


















