第7章：消息代理 Broker 入门——RabbitMQ 与 Redis

0. 上一章思考题参考答案思考题 1消息内容基本相同任务名 args/kwargs 选项差别在投递入口delay走任务对象会用任务级选项queue、serializer 等补全消息send_task按字符串投递直接用全局配置与调用时选项不知道任务定义的默认值。send_task 更解耦——调用方不需要 import 任务模块跨语言系统如 Go 服务按 AMQP 协议直接发也能投递也更危险——任务名拼错没人拦参数错误要到 Worker 执行时才暴露。思考题 2加参数的黄金法则① 新参数一律带默认值老调用方无感② 只用关键字参数、位置参数不超过 2 个加位置参数会挤爆所有现有调用③ 语义变化比如短信从「单发」变「批量」不改原任务新开v2任务名 灰度迁移④ 契约表同步更新并在消费侧做版本字段。1. 项目背景订单短信跑了三周老板问了一个问题「大促时消息能存多少崩了丢不丢」团队一盘点冷汗直流现在的 Broker 是开发机上的单机 Redis没开持久化没有监控没人知道队列有多深。更要命的是上周机房例行重启Redis 进程一重启内存里的 3 万条待发短信直接蒸发——运营又加班手动补发。架构评审会上吵成一团开发组说「Redis 现成继续用」运维组坚持「RabbitMQ 有管理台、能落盘、有确认必须换」架构师问「SQS 呢」谁也说服不了谁。争论的根源是没有一套选型标准——大家不知道 Broker 到底承担什么职责自然各说各话。Broker 的五个职责一个都不能少 ① 解耦生产者写完就走消费者随时来取 ② 削峰流量洪峰先「存」在队列里消费端按能力消化 ③ 持久化进程崩溃/重启消息还在 ④ 确认消费成功与否双方要有凭据Ack ⑤ 路由消息按规则进到该进的队列本章就把 Broker 的职责拆开用同一套任务代码分别对接 RabbitMQ 和 Redis实测两者的差异产出一张选型决策表——以后谁再吵看表说话。2. 项目设计场景选型会上三方僵持大师上台画了两张图。小胖我就纳闷了队列不就是个 List 吗Redis 一个LPUSHBRPOP就完事了几行代码的事你们非得上 RabbitMQ 那一整套「交换机、队列、绑定」跟办个婚礼一样复杂。小白我看过 RabbitMQ 的文档它叫 AMQP 协议消息要先发到交换机Exchange再按routing_key进队列Queue消费者订阅队列。这个「中间加一层交换机」到底图什么Redis 是直接操作 List简单粗暴两者语义差异在哪大师小白画出了要害。RabbitMQ 的模型是「生产者 → Exchange →按绑定规则→ Queue → 消费者」交换机这层让你把「消息怎么走」从业务代码里抽出来——改路由不用改代码改绑定就行还支持 fanout广播、direct精确、topic模式匹配三种规则。Redis 的 List 就是裸队列没有路由、没有消息级别的确认、没有死信全靠 Celery/Kombu 在上层打补丁。再看可靠性RabbitMQ 消息可持久化落盘消费端Ack 确认后才从队列移除Redis 默认内存态靠visibility_timeout可见性超时模拟确认——消息被取走后若 N 秒内没「确认」就重新对其他 Worker 可见。一个是「收据制」一个是「闹钟制」。技术映射RabbitMQ 有传菜系统的大酒楼点单→传菜→划单都有凭据Redis List 自助餐厅的长条桌菜摆上去自己拿掉了没人管超时就重新上。小白可见性超时这个我要追问。我们短信任务可能跑 2 秒但批量导出要跑 10 分钟。如果可见性超时只有 60 秒长任务会不会被当成「死亡」重复投递这跟第 1 章说的「会不会丢、会不会重」直接相关。大师问得太到位了。这就是 Redis Broker 最经典的坑长任务时长必须小于可见性超时否则任务执行到一半消息重新可见被另一个 Worker 拿走再跑一遍——短信发两条。要么把visibility_timeout调大治标期间真崩溃的消息要等很久才重投要么长任务换 RabbitMQAck 是执行完才发的配合acks_late语义更干净第 18 章详解。Redis 做 Broker 只适合「短平快」的任务。小胖那到底选谁别绕弯子。大师给你一张决策表开发环境用 Redis一条 docker 命令起零配置图快生产大流量、任务关键、要管理台 → RabbitMQ在云上、不想管中间件 → SQS 但接受语义弱。反例也要记住中小团队流量小、任务不关键比如发通知生产用 Redis 开启持久化也完全合理——选型看约束不看逼格。我们公司「订单短信对账」全是关键链路生产定 RabbitMQ但开发继续 Redis 过渡。技术映射选 Broker 就像选快递——同城件低风险随便哪家都行合同和贵重物品关键任务必须顺丰保价持久化确认可查。3. 项目实战3.1 环境准备Docker Compose本仓库docker/docker-compose.yml自带 rabbit redis 服务同一套任务代码通过配置切换 Brokerdockercompose-fdocker/docker-compose.yml up-drabbit redis# rabbitmq:management 镜像自带管理台http://localhost:15672guest/guest3.2 分步实现步骤 1一套任务代码两个 Broker 配置目标任务定义不变只切配置。# order_tasks.pyfromceleryimportCelery appCelery(order_tasks)# 不再写死 brokerapp.config_from_object(configs.broker_redis)# 或 configs.broker_rabbit# configs/broker_redis.pybroker_urlredis://localhost:6379/0broker_transport_options{visibility_timeout:60}# 可见性超时 60 秒# configs/broker_rabbit.pybroker_urlpyamqp://guest:guestlocalhost:5672//broker_connection_retry_on_startupTrue# 启动时 Broker 没起来就重试task_acks_lateTrue# 执行完再确认RabbitMQ 语义干净app.task(nameorders.send_order_sms,bindTrue)defsend_order_sms(self,order_id:int)-bool:importtime time.sleep(2)# 模拟 2 秒网关调用print(f[SMS] 订单{order_id}短信已发出)returnTrue步骤 2分别对接两个 Broker 跑通最小闭环目标验证同一套代码在两个 Broker 上都能工作。# Redis 模式开发celery-Aorder_tasks worker--loglevelinfo--poolsolo celery-Aorder_tasks call orders.send_order_sms--args[1]# RabbitMQ 模式生产候选改 config_from_object(configs.broker_rabbit) 后重启 Worker运行结果文字描述两种模式下 Worker 日志均出现Task orders.send_order_sms[...] received与succeeded in 2.0xxs业务行为一致。步骤 3对比「断线重连」行为目标验证配置项broker_connection_retry_on_startup的价值。# 先不起 RabbitMQ直接启动 WorkerRabbitMQ 模式celery-Aorder_tasks worker--loglevelinfo--poolsolo运行结果文字描述未配置该项时报consumer: Cannot connect to amqp://guest:**127.0.0.1:5672//后 Worker 退出配置后日志持续打印Trying to reconnect...Broker 启动后自动恢复消费——生产环境必须开启。步骤 4对比「消息堆积可观测性」目标直观体验「管理台 vs 命令行」。# RabbitMQ管理台 http://localhost:15672 看 Queues 页签ready/unacked 数、消费速率# 或命令行dockerexecdocker-rabbit-1 rabbitmqctl list_queues name messages_ready messages_unacknowledged# Redis命令行看积压dockerexecdocker-redis-1 redis-cli LLEN celery运行结果文字描述RabbitMQ 能同时看到 ready待消费与 unacked已取走未确认两条曲线一眼判断「消费者是否活着」Redis 的 LLEN 只能看到一个数字消息被取走就消失「假性清空」无法感知。步骤 5验证 Redis 可见性超时的「长任务重复」风险目标亲手复现本章最核心的坑可选时长敏感。# long_task.pyfromorder_tasksimportappapp.task(nameorders.long_task,bindTrue)deflong_task(self):importtimeprint(开始执行,self.request.id)time.sleep(90)# 超过 visibility_timeout60print(执行完成,self.request.id)returndone运行文字描述启动 2 个 Worker 消费同一队列投递一次long_task约 60 秒后第二个 Worker 也打印「开始执行」同一条消息重新可见被重复投递。结论长任务在 Redis Broker 上必须调大 visibility_timeout 或换 RabbitMQ。步骤 6亲手摸一摸 AMQP 的「交换机 → 绑定 → 队列」目标把第 2 节的 AMQP 模型落到实体上理解「路由在 Broker 里不在代码里」。# Celery 默认声明的交换机direct 类型名为 celerydockerexecdocker-rabbit-1 rabbitmqctl list_exchanges nametypedockerexecdocker-rabbit-1 rabbitmqctl list_bindingsdockerexecdocker-rabbit-1 rabbitmqctl list_queues name durable运行结果文字描述能看到名为celery的 direct 交换机、celery队列以及一条绑定记录exchangecelery → queueceleryrouting_keycelery。你发一条任务消息本质上就是「带着 routing_key 扔进这个交换机交换机按绑定表把它投进队列」——第 9 章拆队列拆的就是这三样实体的排列组合。3.3 可能遇到的坑及解决方法坑现象解决RabbitMQ 连不上ACCESS_REFUSEDguest 用户只允许本机新建业务账号最小权限只给队列Worker 启动即退Cannot connectBroker 未就绪broker_connection_retry_on_startupTrueRedis 任务「假性清空」LLEN0 但任务没人执行消息被取走未确认看unackedRabbitMQ或调大 visibility_timeout消息被重复执行长任务超过可见性超时任务耗时 visibility_timeout 时换 RabbitMQ 或调大超时RabbitMQ 内存告警堆积过多且消息不落盘设置队列x-max-length或消费端限速第 21 章3.4 完整代码清单与测试验证清单order_tasks.py任务configs/broker_redis.pyconfigs/broker_rabbit.pylong_task.py。选型决策表沉淀 Wiki维度RedisRabbitMQ路由能力无裸队列Exchange/Bindingdirect/fanout/topic可靠性内存态为主可见性超时模拟确认持久化 消息级 Ack可观测LLEN 等有限命令管理台 指标完善运维成本低一条命令中账号/权限/集群适用任务短任务、非关键、开发环境长任务、关键链路、生产环境测试验证# tests/test_broker.pyfromorder_tasksimportappdeftest_broker_switching_is_config_only():# 任务定义与 broker 解耦改配置即可切换assertapp.conf.broker_urlin(redis://localhost:6379/0,pyamqp://guest:guestlocalhost:5672//,)deftest_rabbit_config_has_late_ack():importconfigs.broker_rabbitascassertc.task_acks_lateisTruepython-mpytest tests/test_broker.py-v# 2 passed4. 项目总结4.1 优点 缺点维度Redis 当 BrokerRabbitMQ 当 Broker上手/部署极简一条命令需镜像、账号、权限规划可靠性弱可见性超时、易重复强持久化、Ack路由无Exchange/Binding 三模式可观测弱管理台 指标性能低延迟极快快略逊多语言生态通用AMQP 协议标准生态广4.2 适用场景Redis① 开发/测试环境② 短任务、可容忍偶尔重复的通知类业务③ 已有 Redis 运维体系且不想加中间件的中小团队。RabbitMQ① 生产关键链路订单、支付回调② 长任务分钟级③ 需要消息级确认与死信队列④ 需要管理台与多团队协作。不适用Redis 不适合长任务与强一致投递RabbitMQ 不适合「临时起意跑个 demo」的轻量场景。另外提醒Broker 选型不是一次性决策——业务量从每天 1 万涨到每天 100 万任务时要按本章决策表重新评审别把开发期的临时选型「熬」成生产期的技术债。4.3 注意事项Redis 做 Broker 时visibility_timeout必须 最长任务耗时开启 AOF/RDB 持久化内存上限设maxmemory防 OOM。RabbitMQguest 账号禁用于生产broker_connection_retry_on_startup生产必开队列设置长度上限防堆积打爆磁盘。两者混用是大忌同一业务消息流同时进两个 Broker语义不一致排查地狱。迁移 Broker 时Redis → RabbitMQ要做「双读窗口」新旧 Worker 各部署一套消息只走新 Broker旧队列消化完再下线——直接硬切必然丢在途消息。4.4 常见踩坑经验3 个生产故障故障机房重启后短信消息全部丢失。根因Redis 没开持久化。对策生产 Broker 一律开启 AOF或换 RabbitMQ。教训Broker 的持久化策略是上线前检查项不是可选配。故障RabbitMQ 磁盘写满集群停止投递。根因报表任务堆积几百万条消息都落盘。对策队列设x-max-length 消费端限速 磁盘水位告警。教训Broker 不是无限容量堆积必须有上限。故障订单通知重复推送。根因Redis Broker 可见性超时 60 秒高峰期任务排队超过 60 秒同一条消息被两个 Worker 都取到。对策短任务也要考虑「排队时长执行时长」上限不能只看执行时长。教训可见性超时算的是「取走后到确认」的全周期。4.5 思考题Redis 的可见性超时visibility_timeout和 RabbitMQ 的 Ack 机制在「Worker 崩溃」场景下的行为差异是什么哪个更容易重复执行哪个更容易丢消息为什么说「Exchange/Binding 把路由从代码里抽出来」如果不用交换机模型要在 Redis 上实现「按业务分队列」生产者代码要怎么改答案见第 8 章开头的「上一章思考题参考答案」。延伸阅读与资源Java 工程师进阶从 JVM 生产排障到OpenJDK原理NumPy 从入门到生产落地全链路实战指南科学计算/向量化Redis 8 实战精讲从 CRUD 到源码构建高可用缓存系统Redis 实战修炼与原理进阶Python 3实战精进从脚本到高并发订单引擎python入门Rquests从菜鸟脚本到企业级SDK的网络实战圣经Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地MongoDB 实战进阶与内核修炼后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析