文章摘要本文面向 RabbitMQ 初学者从消息队列的使用场景讲起系统介绍 RabbitMQ 架构、四种交换机、六种常见工作模式并通过 Docker、Python 和 Spring Boot 代码演示消息收发最后讲解 ACK、发布确认、消息持久化、幂等性、死信队列、TTL、Quorum Queue 等生产环境重点。适用读者Java / Spring Boot 后端开发者、微服务学习者、消息中间件初学者。版本说明文章以 RabbitMQ 4.x 为背景涉及 4.3 专属能力时会特别标注。不同版本与客户端配置可能略有差异。一、RabbitMQ 是什么为什么要使用消息队列RabbitMQ 是一个开源的消息代理Message Broker负责接收、路由、存储和投递消息。可以把它想象成快递中转站生产者寄送包裹队列临时保管包裹消费者领取并处理包裹。以电商订单系统为例。用户支付成功后业务可能还需要更新订单状态扣减库存增加积分发送短信或邮件。如果所有服务都由订单服务同步调用接口的总耗时可能增加而且下游服务故障也可能影响主流程。引入 RabbitMQ 后可以让已完成关键事务的订单服务发布一条事件由不同业务服务各自消费订单服务 | 发布订单事件 | RabbitMQ | 交换机Exchange / | \ / | \ 库存队列 积分队列 通知队列 | | | 库存服务 积分服务 通知服务RabbitMQ 的三个常见价值异步把不需要在用户请求中完成的任务放到后台处理解耦生产者只发布事件不需要直接依赖所有下游系统削峰在消息量短时突增时队列缓冲部分压力消费者按自身能力处理。需要注意消息队列不是无限容量的存储器。如果生产速度长期大于消费速度消息仍然会堆积。支付成功、数据库事务提交等关键步骤也不能因为使用了 MQ 就随意改成异步。二、RabbitMQ 的核心架构理解 RabbitMQ先认识六个概念。概念中文名称主要职责Producer生产者发布消息Exchange交换机根据规则将消息路由到队列Queue队列保存等待消费的消息Binding绑定建立交换机与队列之间的关系Routing Key路由键为路由提供匹配信息Consumer消费者接收消息并执行业务消息基本流向Producer | | 发布消息带 Routing Key v Exchange | | 根据 Binding 路由 v Queue | | 投递消息 v Consumer | | 业务处理成功后 ACK v RabbitMQ 确认该次投递已完成重点记住三句话Exchange 决定消息发往哪些队列。Queue 保存尚待处理的消息。Consumer 执行真正的业务逻辑。生产者实际上总是向交换机发布消息。即便使用exchange也是在使用 RabbitMQ 的默认交换机此时可用队列名称作为 Routing Key。RabbitMQ 客户端还经常涉及两个概念Connection客户端与 Broker 的网络连接通常是 TCP 长连接。Channel复用在 Connection 上的逻辑信道执行发布和消费等操作。实际项目通常复用连接及客户端管理的信道而不是每发一条消息就创建新 TCP 连接。加强理解Exchange交换机、Binding绑定关系、Routing Key路由键三者之间的关系可以简单理解为Exchange负责接收消息并决定消息发往哪个队列。Binding建立 Exchange 和 Queue 之间的绑定关系相当于一条路由规则。Routing Key生产者发送消息时指定的路由标识Exchange 根据它和绑定规则决定消息的去向。关系图例子使用的是 Direct ExchangeRouting Key 是消息携带的Binding Key 是绑定关系中配置的。 两者不一定相同是否匹配由 Exchange 类型决定。三、四种交换机Exchange3.1 Direct精确匹配Direct Exchange 使用 Binding Key 与消息 Routing Key 进行精确匹配。Direct Exchange / \ keyerror / \ keyinfo v v error.queue info.queue例如channel.basic_publish(exchangelog.exchange,routing_keyerror,body数据库连接失败.encode(utf-8))如果error.queue绑定了error它就会接收这条消息。多个队列也可以绑定同一个 Key所以 Direct 并不意味着只能路由到一个队列。3.2 Fanout广播Fanout Exchange 忽略 Routing Key将消息复制给所有绑定的队列。Fanout Exchange / | \ v v v 邮件队列 积分队列 统计队列适合“用户注册成功后同时通知多个业务系统”等事件广播场景。3.3 Topic通配符匹配Topic Exchange 根据由点号分隔的 Routing Key 进行模式匹配*匹配一个单词#匹配零个或多个单词。绑定规则消息 Routing Key是否匹配order.*order.created是order.*order.pay.success否order.#order.pay.success是order.#order是user.*order.created否例如订单服务订阅order.#即可匹配订单创建、支付、取消等多类事件。3.4 Headers按消息头匹配Headers Exchange 不依赖 Routing Key而是按消息头字段匹配。可以用x-matchall全部满足或x-matchany满足任意条件控制路由。一般入门优先掌握 Direct、Fanout 和 Topic。四、六种常见消息工作模式不少教程会将 RabbitMQ 的入门使用方式分成六种模式核心特点典型场景Simple一个生产者、一个队列、一个消费者最简单的消息收发Work Queues多个消费者竞争同一个队列后台任务分摊Publish/SubscribeFanout 广播到多个队列用户注册通知RoutingDirect 精确路由按日志级别分流TopicsTopic 通配符路由订单事件订阅RPC请求消息与响应消息配合远程调用式交互这里要特别区分一个队列多个消费者一般是竞争消费同一条投递消息通常由其中一个消费者处理。多个队列绑定同一个交换机每个匹配的队列都可以获得该消息的一份副本。五、Docker Python搭建第一个可运行示例下面以订单创建事件为例使用一个 Direct Exchange 和一个持久化队列。5.1 启动 RabbitMQ确保已安装 Docker然后运行dockerrun-d\--namerabbitmq-study\--hostnamerabbitmq-study\-p5672:5672\-p15672:15672\-eRABBITMQ_DEFAULT_USERdemo\-eRABBITMQ_DEFAULT_PASSdemo123\rabbitmq:4-management端口说明5672AMQP 客户端连接端口15672RabbitMQ Web 管理后台。浏览器访问http://localhost:15672使用demo / demo123登录。以上凭据仅用于本地学习生产环境应使用安全凭据与合理的网络访问控制。上线时也建议固定 Docker 镜像的具体版本。5.2 安装客户端python-mpipinstallpika5.3 编写消费者receive.pyimportjsonimportpika credentialspika.PlainCredentials(demo,demo123)connectionpika.BlockingConnection(pika.ConnectionParameters(hostlocalhost,credentialscredentials))channelconnection.channel()# 声明交换机、队列和绑定channel.exchange_declare(exchangeorder.events,exchange_typedirect,durableTrue)channel.queue_declare(queueorder.created.queue,durableTrue)channel.queue_bind(exchangeorder.events,queueorder.created.queue,routing_keyorder.created)# 每个消费者最多持有一条尚未确认的消息channel.basic_qos(prefetch_count1)defcallback(ch,method,properties,body):try:orderjson.loads(body.decode(utf-8))print(收到订单,order)# TODO: 在这里编写业务处理逻辑print(处理完成)ch.basic_ack(delivery_tagmethod.delivery_tag)exceptExceptionasexc:print(处理失败,exc)# 演示重新入队。生产环境必须限制重试避免死循环。ch.basic_nack(delivery_tagmethod.delivery_tag,requeueTrue)channel.basic_consume(queueorder.created.queue,on_message_callbackcallback,auto_ackFalse)print(等待订单消息...)channel.start_consuming()5.4 编写生产者send.pyimportjsonimportuuidimportpika credentialspika.PlainCredentials(demo,demo123)connectionpika.BlockingConnection(pika.ConnectionParameters(hostlocalhost,credentialscredentials))channelconnection.channel()channel.exchange_declare(exchangeorder.events,exchange_typedirect,durableTrue)channel.queue_declare(queueorder.created.queue,durableTrue)channel.queue_bind(exchangeorder.events,queueorder.created.queue,routing_keyorder.created)# 发布者确认模式channel.confirm_delivery()order{orderId:1001,userId:2001,amount:99.9}channel.basic_publish(exchangeorder.events,routing_keyorder.created,bodyjson.dumps(order).encode(utf-8),mandatoryTrue,propertiespika.BasicProperties(delivery_mode2,content_typeapplication/json,message_idstr(uuid.uuid4())))print(消息发布已被 RabbitMQ 确认)connection.close()代码中durableTrue使队列/交换机定义在 Broker 重启后保留delivery_mode2将消息标记为持久化confirm_delivery()让发布者等待 Broker 确认mandatoryTrue消息无法路由时要求 Broker 返回Pika 的确认模式可以据此发现不可路由问题auto_ackFalse采用消费端手动确认。5.5 运行测试打开两个终端。终端一python receive.py终端二python send.py消费者将打印类似结果等待订单消息... 收到订单 {orderId: 1001, userId: 2001, amount: 99.9} 处理完成可以在 RabbitMQ 管理后台查看 Exchanges、Queues、Connections进一步理解消息流转。六、消息可靠性生产环境最重要的部分6.1 发布者确认Publisher Confirm生产者写入网络不代表 Broker 一定收到消息。开启 Publisher Confirm 后Broker 会对发布结果进行确认。Producer --发送消息-- RabbitMQ Producer --Confirm---- RabbitMQ注意Publisher Confirm 不表示消费者已完成业务处理。它解决的是生产者到 Broker 的发布确认问题。即使收到 Confirm ACK也应该检查消息是否成功路由到预期队列无法路由时可配合mandatory返回机制处理。6.2 消费者确认Consumer ACK消费者接收到消息后不应直接等同于“业务完成”。手动 ACK 的推荐流程是收到消息 - 执行业务 - 数据库事务提交成功 - 发送 basic_ackAPI含义basic_ack确认消息处理成功basic_nack(requeueTrue)拒绝并重新入队basic_nack(requeueFalse)拒绝且不重新入队basic_reject拒绝单条消息若配置了死信交换机符合条件的拒绝消息可转入死信处理否则不重新入队的消息可能被丢弃。6.3 消息持久化要分别考虑交换机是否持久化队列是否持久化消息本身是否被标记为持久化是否使用发布确认判断 Broker 已安全接管消息。“持久化”并不意味着绝对零丢失。磁盘故障、集群配置、发布确认以及队列类型都会影响实际的数据安全性。6.4 为什么还需要幂等性例如积分消费者已经修改数据库但在发送 ACK 前宕机消息到达 - 增加积分成功 - 消费者宕机 - ACK 未到达 | v Broker 重新投递消费者可能重复处理同一条消息。因此重要业务需要保证幂等即重复处理不会重复产生业务效果。常见做法消息包含唯一messageId或业务orderId使用唯一约束和数据库事务去重。例如CREATETABLEprocessed_message(message_idVARCHAR(100)PRIMARYKEY);实际处理时在同一个数据库事务中完成“插入去重记录”和“修改积分”。如果唯一键已存在就跳过重复业务操作。只做“先查询是否处理过”还不够因为并发消费者可能同时查到未处理。6.5 数据库与 MQ 的双写一致性考虑下面的情况订单数据库提交成功 - 发送 MQ 消息失败 - 下游没有收到订单事件一种常见解决方案是Transactional Outbox事务性发件箱在同一个数据库事务中写入业务数据和待发布事件由后台任务发布事件并在获得确认后更新发送状态。该方案仍可能重复发布因此消费端幂等不能省略。七、死信队列、TTL、延迟处理与重试7.1 死信队列DLQ消息被拒绝且不重新入队、到期过期、超出队列长度限制等情况下消息可能成为死信。业务队列 | | 被拒绝 / 过期 / 超限 v 死信交换机DLX | v 死信队列DLQ | v 排查异常、补偿、人工处理DLQ 通常是自己声明的普通队列通过 DLX 及 Binding 来接收死信不是一个自动生成的特殊队列类型。7.2 TTL消息生存时间例如channel.queue_declare(queuetemporary.queue,durableTrue,arguments{x-message-ttl:30000})30000的单位是毫秒表示消息在队列中的生存时间为 30 秒。到期后消息不再作为正常消息交付若正确配置 DLX可以进一步做死信转发。TTL 不等于高精度定时任务。7.3 延迟消息与延迟重试不是同一件事典型延迟业务订单创建 30 分钟后检查支付状态仍未支付则关闭订单。传统实现常采用 TTL DLX 等方案但它不是精确调度器。版本补充RabbitMQ 4.3 的 Quorum Queue 支持延迟重试可以对返回队列的消息设置延迟和线性退避减少反复立即重新投递带来的压力。该能力针对失败后重投不应直接等同于通用定时任务。旧版本不能默认使用这一能力。无论使用哪种重试方式都应考虑最大重试次数、重试间隔、死信兜底和告警避免无限重试。八、Prefetch控制消费者的未确认消息数量channel.basic_qos(prefetch_count10)常见的消费设置下表示单个消费者最多可同时持有 10 条未确认消息。太小吞吐可能受限太大消费者可能积压大量尚未确认的消息合适的值根据耗时、并发和消息大小进行压测。在初学示例中设置为1很直观但不一定适合高吞吐量生产场景。九、Classic Queue、Quorum Queue 和 Stream类型主要特点适合场景Classic Queue传统消息队列配置与使用相对简单常规消息任务Quorum Queue基于 Raft 的复制队列强调数据安全和高可用关键订单、重要事件Stream面向追加式日志和消息重放高吞吐事件流、历史事件回放Quorum Queue 声明示例channel.queue_declare(queuecritical.orders,durableTrue,arguments{x-queue-type:quorum})Quorum Queue 的跨节点高可用需要真正的多节点部署和正确的副本配置。只在一台机器上声明 Quorum Queue 并不会自动实现跨节点容灾。十、Spring Boot 整合 RabbitMQ10.1 添加依赖dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency10.2 application.ymlspring:rabbitmq:host:localhostport:5672username:demopassword:demo12310.3 声明交换机、队列和绑定ConfigurationpublicclassRabbitConfig{BeanpublicDirectExchangeorderExchange(){returnnewDirectExchange(order.events,true,false);}BeanpublicQueueorderQueue(){returnnewQueue(order.created.queue,true);}BeanpublicBindingorderBinding(QueueorderQueue,DirectExchangeorderExchange){returnBindingBuilder.bind(orderQueue).to(orderExchange).with(order.created);}}所需Configuration、Bean等注解及Queue、BindingBuilder等类可通过 IDE 自动导入对应的 Spring 和 Spring AMQP 包。10.4 生产者ServicepublicclassOrderProducer{privatefinalRabbitTemplaterabbitTemplate;publicOrderProducer(RabbitTemplaterabbitTemplate){this.rabbitTemplaterabbitTemplate;}publicvoidsendOrderMessage(StringorderId){rabbitTemplate.convertAndSend(order.events,order.created,orderId);}}10.5 消费者ComponentpublicclassOrderConsumer{RabbitListener(queuesorder.created.queue)publicvoidreceive(StringorderId){System.out.println(收到订单orderId);// TODO: 执行业务逻辑}}注意一个容易踩的坑Spring AMQP 的AUTO确认模式并不是 AMQP 协议层的auto_acktrue。在 Spring AMQPAUTO模式下监听容器通常在方法正常完成后确认消息如果希望由业务代码明确调用basicAck可配置MANUAL并处理异常场景。上面是最基础的收发示例尚未包含生产必需的发布确认、异常重试、幂等和数据库一致性处理。十一、RabbitMQ 和 Kafka 的区别维度RabbitMQKafka核心定位通用消息代理分布式事件流平台路由方式Exchange Binding Routing KeyTopic Partition常用模式任务队列、竞争消费、广播和复杂路由事件日志、消费组、分区处理消息重放普通队列不是主要面向重放设计Stream 支持消息重放是核心能力常见场景异步任务、业务事件分发、工作队列事件流、日志采集、数据管道可以粗略理解RabbitMQ 更注重消息路由与任务分发Kafka 更注重持久化事件日志与重放。选型还要看吞吐量、消息保留、可靠性和团队技术栈。十二、生产环境检查清单检查项建议发布可靠性开启发布确认处理不可路由消息消费可靠性使用恰当的 ACK 策略消息安全按需设置持久化和复制队列重复消费通过数据库唯一键等手段保证幂等异常处理有限重试、退避、死信和告警消息积压监控 Ready、Unacked 和消费吞吐并发控制调整消费者数量与 Prefetch数据一致性重要业务考虑事务性 Outbox安全配置使用强密码、权限隔离、必要时启用 TLS管理后台中重点关注Ready已在队列中、等待交付的消息Unacked已经投递但尚未确认的消息Total队列中的消息总量通常包含 Ready 与 Unacked。Ready 长期增长通常意味着生产速度高于消费速度Unacked 长期偏高可能说明消费者处理慢、阻塞或确认逻辑有问题。十三、常见面试问题Q1消息一定会被消费一次吗不一定。存在网络失败、重投和 ACK 丢失等情况常见可靠消费设计通常需要考虑至少一次投递带来的重复处理因此业务幂等非常重要。Q2Publisher Confirm 与 Consumer ACK 有什么区别Confirm 表示 Broker 对生产者发布的消息作出确认Consumer ACK 表示消费者对 Broker 的投递作出确认。两者属于不同阶段。Q3一个队列绑定多个消费者会广播吗通常不会。它属于竞争消费一条投递消息一般由其中一个消费者处理。广播需要多条队列分别绑定交换机。Q4持久化队列就不会丢消息了吗不能这样理解。还需结合持久化消息、发布确认、磁盘与集群容灾等因素判断。Q5RabbitMQ 消息积压怎么办先排查消费者故障和下游数据库瓶颈再根据业务能力调整消费者并发、Prefetch 或扩容同时防止无限堆积必要时采用限流、降级与告警。十四、学习路线总结建议按以下顺序循序渐进基础概念Producer、Exchange、Queue、Binding、Routing Key消息模式Direct、Fanout、Topic 和竞争消费代码实战Docker 启动、Python 或 Spring Boot 收发消息可靠性设计发布确认、手动 ACK、消息持久化和幂等高级功能DLX、TTL、重试、Prefetch、Quorum Queue项目实践设计完整的订单事件流并考虑数据库事务一致性、监控和故障恢复。一句话总结RabbitMQ 通过交换机决定消息如何路由通过队列承接消息通过消费者异步处理业务再利用确认、持久化、重试、死信和幂等机制提升可靠性。参考资料RabbitMQ 官方文档https://www.rabbitmq.com/docsRabbitMQ 官方教程https://www.rabbitmq.com/tutorialsRabbitMQ 发布确认与消费确认https://www.rabbitmq.com/docs/confirmsRabbitMQ Quorum Queuehttps://www.rabbitmq.com/docs/quorum-queuesSpring AMQP 文档https://docs.spring.io/spring-amqp/reference/作者笔记本文适合先收藏、边阅读边运行示例。学习 RabbitMQ 时建议不仅跑通“发送成功”还要亲自模拟消费者异常、消息重试和重复消费才能真正理解可靠性机制。 
长河君 · 2026/10/14 9:33:10 · 阅读 1.2k RabbitMQ 从入门到实战：核心原理、消息模型、可靠性机制与 Spring Boot 实战
文章摘要本文面向 RabbitMQ 初学者从消息队列的使用场景讲起系统介绍 RabbitMQ 架构、四种交换机、六种常见工作模式并通过 Docker、Python 和 Spring Boot 代码演示消息收发最后讲解 ACK、发布确认、消息持久化、幂等性、死信队列、TTL、Quorum Queue 等生产环境重点。适用读者Java / Spring Boot 后端开发者、微服务学习者、消息中间件初学者。版本说明文章以 RabbitMQ 4.x 为背景涉及 4.3 专属能力时会特别标注。不同版本与客户端配置可能略有差异。一、RabbitMQ 是什么为什么要使用消息队列RabbitMQ 是一个开源的消息代理Message Broker负责接收、路由、存储和投递消息。可以把它想象成快递中转站生产者寄送包裹队列临时保管包裹消费者领取并处理包裹。以电商订单系统为例。用户支付成功后业务可能还需要更新订单状态扣减库存增加积分发送短信或邮件。如果所有服务都由订单服务同步调用接口的总耗时可能增加而且下游服务故障也可能影响主流程。引入 RabbitMQ 后可以让已完成关键事务的订单服务发布一条事件由不同业务服务各自消费订单服务 | 发布订单事件 | RabbitMQ | 交换机Exchange / | \ / | \ 库存队列 积分队列 通知队列 | | | 库存服务 积分服务 通知服务RabbitMQ 的三个常见价值异步把不需要在用户请求中完成的任务放到后台处理解耦生产者只发布事件不需要直接依赖所有下游系统削峰在消息量短时突增时队列缓冲部分压力消费者按自身能力处理。需要注意消息队列不是无限容量的存储器。如果生产速度长期大于消费速度消息仍然会堆积。支付成功、数据库事务提交等关键步骤也不能因为使用了 MQ 就随意改成异步。二、RabbitMQ 的核心架构理解 RabbitMQ先认识六个概念。概念中文名称主要职责Producer生产者发布消息Exchange交换机根据规则将消息路由到队列Queue队列保存等待消费的消息Binding绑定建立交换机与队列之间的关系Routing Key路由键为路由提供匹配信息Consumer消费者接收消息并执行业务消息基本流向Producer | | 发布消息带 Routing Key v Exchange | | 根据 Binding 路由 v Queue | | 投递消息 v Consumer | | 业务处理成功后 ACK v RabbitMQ 确认该次投递已完成重点记住三句话Exchange 决定消息发往哪些队列。Queue 保存尚待处理的消息。Consumer 执行真正的业务逻辑。生产者实际上总是向交换机发布消息。即便使用exchange也是在使用 RabbitMQ 的默认交换机此时可用队列名称作为 Routing Key。RabbitMQ 客户端还经常涉及两个概念Connection客户端与 Broker 的网络连接通常是 TCP 长连接。Channel复用在 Connection 上的逻辑信道执行发布和消费等操作。实际项目通常复用连接及客户端管理的信道而不是每发一条消息就创建新 TCP 连接。加强理解Exchange交换机、Binding绑定关系、Routing Key路由键三者之间的关系可以简单理解为Exchange负责接收消息并决定消息发往哪个队列。Binding建立 Exchange 和 Queue 之间的绑定关系相当于一条路由规则。Routing Key生产者发送消息时指定的路由标识Exchange 根据它和绑定规则决定消息的去向。关系图例子使用的是 Direct ExchangeRouting Key 是消息携带的Binding Key 是绑定关系中配置的。 两者不一定相同是否匹配由 Exchange 类型决定。三、四种交换机Exchange3.1 Direct精确匹配Direct Exchange 使用 Binding Key 与消息 Routing Key 进行精确匹配。Direct Exchange / \ keyerror / \ keyinfo v v error.queue info.queue例如channel.basic_publish(exchangelog.exchange,routing_keyerror,body数据库连接失败.encode(utf-8))如果error.queue绑定了error它就会接收这条消息。多个队列也可以绑定同一个 Key所以 Direct 并不意味着只能路由到一个队列。3.2 Fanout广播Fanout Exchange 忽略 Routing Key将消息复制给所有绑定的队列。Fanout Exchange / | \ v v v 邮件队列 积分队列 统计队列适合“用户注册成功后同时通知多个业务系统”等事件广播场景。3.3 Topic通配符匹配Topic Exchange 根据由点号分隔的 Routing Key 进行模式匹配*匹配一个单词#匹配零个或多个单词。绑定规则消息 Routing Key是否匹配order.*order.created是order.*order.pay.success否order.#order.pay.success是order.#order是user.*order.created否例如订单服务订阅order.#即可匹配订单创建、支付、取消等多类事件。3.4 Headers按消息头匹配Headers Exchange 不依赖 Routing Key而是按消息头字段匹配。可以用x-matchall全部满足或x-matchany满足任意条件控制路由。一般入门优先掌握 Direct、Fanout 和 Topic。四、六种常见消息工作模式不少教程会将 RabbitMQ 的入门使用方式分成六种模式核心特点典型场景Simple一个生产者、一个队列、一个消费者最简单的消息收发Work Queues多个消费者竞争同一个队列后台任务分摊Publish/SubscribeFanout 广播到多个队列用户注册通知RoutingDirect 精确路由按日志级别分流TopicsTopic 通配符路由订单事件订阅RPC请求消息与响应消息配合远程调用式交互这里要特别区分一个队列多个消费者一般是竞争消费同一条投递消息通常由其中一个消费者处理。多个队列绑定同一个交换机每个匹配的队列都可以获得该消息的一份副本。五、Docker Python搭建第一个可运行示例下面以订单创建事件为例使用一个 Direct Exchange 和一个持久化队列。5.1 启动 RabbitMQ确保已安装 Docker然后运行dockerrun-d\--namerabbitmq-study\--hostnamerabbitmq-study\-p5672:5672\-p15672:15672\-eRABBITMQ_DEFAULT_USERdemo\-eRABBITMQ_DEFAULT_PASSdemo123\rabbitmq:4-management端口说明5672AMQP 客户端连接端口15672RabbitMQ Web 管理后台。浏览器访问http://localhost:15672使用demo / demo123登录。以上凭据仅用于本地学习生产环境应使用安全凭据与合理的网络访问控制。上线时也建议固定 Docker 镜像的具体版本。5.2 安装客户端python-mpipinstallpika5.3 编写消费者receive.pyimportjsonimportpika credentialspika.PlainCredentials(demo,demo123)connectionpika.BlockingConnection(pika.ConnectionParameters(hostlocalhost,credentialscredentials))channelconnection.channel()# 声明交换机、队列和绑定channel.exchange_declare(exchangeorder.events,exchange_typedirect,durableTrue)channel.queue_declare(queueorder.created.queue,durableTrue)channel.queue_bind(exchangeorder.events,queueorder.created.queue,routing_keyorder.created)# 每个消费者最多持有一条尚未确认的消息channel.basic_qos(prefetch_count1)defcallback(ch,method,properties,body):try:orderjson.loads(body.decode(utf-8))print(收到订单,order)# TODO: 在这里编写业务处理逻辑print(处理完成)ch.basic_ack(delivery_tagmethod.delivery_tag)exceptExceptionasexc:print(处理失败,exc)# 演示重新入队。生产环境必须限制重试避免死循环。ch.basic_nack(delivery_tagmethod.delivery_tag,requeueTrue)channel.basic_consume(queueorder.created.queue,on_message_callbackcallback,auto_ackFalse)print(等待订单消息...)channel.start_consuming()5.4 编写生产者send.pyimportjsonimportuuidimportpika credentialspika.PlainCredentials(demo,demo123)connectionpika.BlockingConnection(pika.ConnectionParameters(hostlocalhost,credentialscredentials))channelconnection.channel()channel.exchange_declare(exchangeorder.events,exchange_typedirect,durableTrue)channel.queue_declare(queueorder.created.queue,durableTrue)channel.queue_bind(exchangeorder.events,queueorder.created.queue,routing_keyorder.created)# 发布者确认模式channel.confirm_delivery()order{orderId:1001,userId:2001,amount:99.9}channel.basic_publish(exchangeorder.events,routing_keyorder.created,bodyjson.dumps(order).encode(utf-8),mandatoryTrue,propertiespika.BasicProperties(delivery_mode2,content_typeapplication/json,message_idstr(uuid.uuid4())))print(消息发布已被 RabbitMQ 确认)connection.close()代码中durableTrue使队列/交换机定义在 Broker 重启后保留delivery_mode2将消息标记为持久化confirm_delivery()让发布者等待 Broker 确认mandatoryTrue消息无法路由时要求 Broker 返回Pika 的确认模式可以据此发现不可路由问题auto_ackFalse采用消费端手动确认。5.5 运行测试打开两个终端。终端一python receive.py终端二python send.py消费者将打印类似结果等待订单消息... 收到订单 {orderId: 1001, userId: 2001, amount: 99.9} 处理完成可以在 RabbitMQ 管理后台查看 Exchanges、Queues、Connections进一步理解消息流转。六、消息可靠性生产环境最重要的部分6.1 发布者确认Publisher Confirm生产者写入网络不代表 Broker 一定收到消息。开启 Publisher Confirm 后Broker 会对发布结果进行确认。Producer --发送消息-- RabbitMQ Producer --Confirm---- RabbitMQ注意Publisher Confirm 不表示消费者已完成业务处理。它解决的是生产者到 Broker 的发布确认问题。即使收到 Confirm ACK也应该检查消息是否成功路由到预期队列无法路由时可配合mandatory返回机制处理。6.2 消费者确认Consumer ACK消费者接收到消息后不应直接等同于“业务完成”。手动 ACK 的推荐流程是收到消息 - 执行业务 - 数据库事务提交成功 - 发送 basic_ackAPI含义basic_ack确认消息处理成功basic_nack(requeueTrue)拒绝并重新入队basic_nack(requeueFalse)拒绝且不重新入队basic_reject拒绝单条消息若配置了死信交换机符合条件的拒绝消息可转入死信处理否则不重新入队的消息可能被丢弃。6.3 消息持久化要分别考虑交换机是否持久化队列是否持久化消息本身是否被标记为持久化是否使用发布确认判断 Broker 已安全接管消息。“持久化”并不意味着绝对零丢失。磁盘故障、集群配置、发布确认以及队列类型都会影响实际的数据安全性。6.4 为什么还需要幂等性例如积分消费者已经修改数据库但在发送 ACK 前宕机消息到达 - 增加积分成功 - 消费者宕机 - ACK 未到达 | v Broker 重新投递消费者可能重复处理同一条消息。因此重要业务需要保证幂等即重复处理不会重复产生业务效果。常见做法消息包含唯一messageId或业务orderId使用唯一约束和数据库事务去重。例如CREATETABLEprocessed_message(message_idVARCHAR(100)PRIMARYKEY);实际处理时在同一个数据库事务中完成“插入去重记录”和“修改积分”。如果唯一键已存在就跳过重复业务操作。只做“先查询是否处理过”还不够因为并发消费者可能同时查到未处理。6.5 数据库与 MQ 的双写一致性考虑下面的情况订单数据库提交成功 - 发送 MQ 消息失败 - 下游没有收到订单事件一种常见解决方案是Transactional Outbox事务性发件箱在同一个数据库事务中写入业务数据和待发布事件由后台任务发布事件并在获得确认后更新发送状态。该方案仍可能重复发布因此消费端幂等不能省略。七、死信队列、TTL、延迟处理与重试7.1 死信队列DLQ消息被拒绝且不重新入队、到期过期、超出队列长度限制等情况下消息可能成为死信。业务队列 | | 被拒绝 / 过期 / 超限 v 死信交换机DLX | v 死信队列DLQ | v 排查异常、补偿、人工处理DLQ 通常是自己声明的普通队列通过 DLX 及 Binding 来接收死信不是一个自动生成的特殊队列类型。7.2 TTL消息生存时间例如channel.queue_declare(queuetemporary.queue,durableTrue,arguments{x-message-ttl:30000})30000的单位是毫秒表示消息在队列中的生存时间为 30 秒。到期后消息不再作为正常消息交付若正确配置 DLX可以进一步做死信转发。TTL 不等于高精度定时任务。7.3 延迟消息与延迟重试不是同一件事典型延迟业务订单创建 30 分钟后检查支付状态仍未支付则关闭订单。传统实现常采用 TTL DLX 等方案但它不是精确调度器。版本补充RabbitMQ 4.3 的 Quorum Queue 支持延迟重试可以对返回队列的消息设置延迟和线性退避减少反复立即重新投递带来的压力。该能力针对失败后重投不应直接等同于通用定时任务。旧版本不能默认使用这一能力。无论使用哪种重试方式都应考虑最大重试次数、重试间隔、死信兜底和告警避免无限重试。八、Prefetch控制消费者的未确认消息数量channel.basic_qos(prefetch_count10)常见的消费设置下表示单个消费者最多可同时持有 10 条未确认消息。太小吞吐可能受限太大消费者可能积压大量尚未确认的消息合适的值根据耗时、并发和消息大小进行压测。在初学示例中设置为1很直观但不一定适合高吞吐量生产场景。九、Classic Queue、Quorum Queue 和 Stream类型主要特点适合场景Classic Queue传统消息队列配置与使用相对简单常规消息任务Quorum Queue基于 Raft 的复制队列强调数据安全和高可用关键订单、重要事件Stream面向追加式日志和消息重放高吞吐事件流、历史事件回放Quorum Queue 声明示例channel.queue_declare(queuecritical.orders,durableTrue,arguments{x-queue-type:quorum})Quorum Queue 的跨节点高可用需要真正的多节点部署和正确的副本配置。只在一台机器上声明 Quorum Queue 并不会自动实现跨节点容灾。十、Spring Boot 整合 RabbitMQ10.1 添加依赖dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency10.2 application.ymlspring:rabbitmq:host:localhostport:5672username:demopassword:demo12310.3 声明交换机、队列和绑定ConfigurationpublicclassRabbitConfig{BeanpublicDirectExchangeorderExchange(){returnnewDirectExchange(order.events,true,false);}BeanpublicQueueorderQueue(){returnnewQueue(order.created.queue,true);}BeanpublicBindingorderBinding(QueueorderQueue,DirectExchangeorderExchange){returnBindingBuilder.bind(orderQueue).to(orderExchange).with(order.created);}}所需Configuration、Bean等注解及Queue、BindingBuilder等类可通过 IDE 自动导入对应的 Spring 和 Spring AMQP 包。10.4 生产者ServicepublicclassOrderProducer{privatefinalRabbitTemplaterabbitTemplate;publicOrderProducer(RabbitTemplaterabbitTemplate){this.rabbitTemplaterabbitTemplate;}publicvoidsendOrderMessage(StringorderId){rabbitTemplate.convertAndSend(order.events,order.created,orderId);}}10.5 消费者ComponentpublicclassOrderConsumer{RabbitListener(queuesorder.created.queue)publicvoidreceive(StringorderId){System.out.println(收到订单orderId);// TODO: 执行业务逻辑}}注意一个容易踩的坑Spring AMQP 的AUTO确认模式并不是 AMQP 协议层的auto_acktrue。在 Spring AMQPAUTO模式下监听容器通常在方法正常完成后确认消息如果希望由业务代码明确调用basicAck可配置MANUAL并处理异常场景。上面是最基础的收发示例尚未包含生产必需的发布确认、异常重试、幂等和数据库一致性处理。十一、RabbitMQ 和 Kafka 的区别维度RabbitMQKafka核心定位通用消息代理分布式事件流平台路由方式Exchange Binding Routing KeyTopic Partition常用模式任务队列、竞争消费、广播和复杂路由事件日志、消费组、分区处理消息重放普通队列不是主要面向重放设计Stream 支持消息重放是核心能力常见场景异步任务、业务事件分发、工作队列事件流、日志采集、数据管道可以粗略理解RabbitMQ 更注重消息路由与任务分发Kafka 更注重持久化事件日志与重放。选型还要看吞吐量、消息保留、可靠性和团队技术栈。十二、生产环境检查清单检查项建议发布可靠性开启发布确认处理不可路由消息消费可靠性使用恰当的 ACK 策略消息安全按需设置持久化和复制队列重复消费通过数据库唯一键等手段保证幂等异常处理有限重试、退避、死信和告警消息积压监控 Ready、Unacked 和消费吞吐并发控制调整消费者数量与 Prefetch数据一致性重要业务考虑事务性 Outbox安全配置使用强密码、权限隔离、必要时启用 TLS管理后台中重点关注Ready已在队列中、等待交付的消息Unacked已经投递但尚未确认的消息Total队列中的消息总量通常包含 Ready 与 Unacked。Ready 长期增长通常意味着生产速度高于消费速度Unacked 长期偏高可能说明消费者处理慢、阻塞或确认逻辑有问题。十三、常见面试问题Q1消息一定会被消费一次吗不一定。存在网络失败、重投和 ACK 丢失等情况常见可靠消费设计通常需要考虑至少一次投递带来的重复处理因此业务幂等非常重要。Q2Publisher Confirm 与 Consumer ACK 有什么区别Confirm 表示 Broker 对生产者发布的消息作出确认Consumer ACK 表示消费者对 Broker 的投递作出确认。两者属于不同阶段。Q3一个队列绑定多个消费者会广播吗通常不会。它属于竞争消费一条投递消息一般由其中一个消费者处理。广播需要多条队列分别绑定交换机。Q4持久化队列就不会丢消息了吗不能这样理解。还需结合持久化消息、发布确认、磁盘与集群容灾等因素判断。Q5RabbitMQ 消息积压怎么办先排查消费者故障和下游数据库瓶颈再根据业务能力调整消费者并发、Prefetch 或扩容同时防止无限堆积必要时采用限流、降级与告警。十四、学习路线总结建议按以下顺序循序渐进基础概念Producer、Exchange、Queue、Binding、Routing Key消息模式Direct、Fanout、Topic 和竞争消费代码实战Docker 启动、Python 或 Spring Boot 收发消息可靠性设计发布确认、手动 ACK、消息持久化和幂等高级功能DLX、TTL、重试、Prefetch、Quorum Queue项目实践设计完整的订单事件流并考虑数据库事务一致性、监控和故障恢复。一句话总结RabbitMQ 通过交换机决定消息如何路由通过队列承接消息通过消费者异步处理业务再利用确认、持久化、重试、死信和幂等机制提升可靠性。参考资料RabbitMQ 官方文档https://www.rabbitmq.com/docsRabbitMQ 官方教程https://www.rabbitmq.com/tutorialsRabbitMQ 发布确认与消费确认https://www.rabbitmq.com/docs/confirmsRabbitMQ Quorum Queuehttps://www.rabbitmq.com/docs/quorum-queuesSpring AMQP 文档https://docs.spring.io/spring-amqp/reference/作者笔记本文适合先收藏、边阅读边运行示例。学习 RabbitMQ 时建议不仅跑通“发送成功”还要亲自模拟消费者异常、消息重试和重复消费才能真正理解可靠性机制。