Spring Boot 集成 RabbitMQ 这事我从实际项目角度聊聊我的做法和踩过的坑。很多人在预约服务系统、订单通知、异步任务这些场景里选型消息队列最终都落在 RabbitMQ 上——因为它轻量、可靠、路由灵活而且 Spring Boot 对它有非常完善的自动配置。这篇文章我会从整体设计讲起把环境搭建、核心代码、可靠性保障、常见问题排查全部过一遍适合刚准备把 RabbitMQ 用进 Spring Boot 项目的人也适合已经用起来但被重复消费、消息丢失、积压这些问题折腾过的人。1. 项目整体设计与核心思路1.1 为什么是 RabbitMQ 而不是别的消息队列我在不少项目里做过消息中间件选型如果团队技术栈是 Java Spring BootRabbitMQ 往往是最稳的选择。它能解决的核心问题就三个异步、解耦、削峰。拿热搜里那个《基于 Spring Boot 的上门烹饪预约服务系统》来说用户下单后要去通知厨师、发短信、记录日志如果这些全在请求线程里同步做接口响应时间很容易从 200ms 冲到 2s。引入 RabbitMQ 之后下单接口只需要把订单事件丢进队列马上返回“下单成功”后面所有耗时的动作都交给消费者异步去做。这就是异步带来的直接收益。那为什么不选 KafkaKafka 的消息模型是分区追加日志吞吐量极高适合大数据量、日志采集、流处理场景。但它部署依赖 ZooKeeper新版本已经有 KRaft 模式运维成本更高而且 RabbitMQ 在复杂路由规则上的表现更好。在一个预约服务这类中小规模业务里每秒几百上千条消息就顶天了RabbitMQ 完全扛得住没必要引入 Kafka 的运维复杂度。RabbitMQ 另一个很典型的应用是延迟队列比如“用户下单后 30 分钟未支付自动取消”“预约时间临近提醒厨师备菜”这类定时任务需求用死信交换机实现非常优雅不需要自己写 Quartz 轮询扫表。1.2 RabbitMQ 的几个核心概念务必在动手前搞清我见过不少新手在写代码前没搞清楚这些概念后面配置交换机、队列时一头雾水。确实需要先弄明白这几个核心对象生产者Producer发消息的一方在 Spring Boot 里就是注入 RabbitTemplate 的 Service 或 Controller。消费者Consumer接收消息的一方在 Spring Boot 里就是标注了 RabbitListener 的方法。交换机Exchange消息分发的中转站它自己不存消息只负责把带路由键的消息投递到匹配的队列。队列Queue真正存储消息的地方。绑定Binding把交换机和队列关联起来的规则相当于一条定义“什么路由键进什么队列”的连线。虚拟主机Virtual Host可以理解为一个独立的小型 RabbitMQ 实例不同项目之间通过 vhost 隔离。交换机有四种类型直接型Direct、主题型Topic、广播型Fanout、头部型Headers。实际项目里最常用的是 Direct 和 TopicFanout 用于广播场景Headers 基本用不到。我用生活化的类比来解释一下交换机就像一个快递分拨中心队列是各个片区的快递站点。你寄快递时填写的目的地地址就是路由键Routing Key分拨中心根据这个地址把包裹送到对应片区的站点。如果你是高级会员还可以指定“一定要走航空件”这就是绑定规则里的参数。这么一理解Exchange 和 Queue 的关系就清楚了队列负责存交换机负责转绑定规则负责告诉交换机怎么转。1.3 消息队列在预约业务中的典型调用链路实操之前我先画一条典型的业务链路出来。不要依赖 mermaid 图我用文字描述你跟着走一遍就明白了用户在小程序端发起预约 - Spring Boot 的 OrderService 保存订单状态为“待确认” - 同时调用 rabbitTemplate.convertAndSend(order.exchange, order.created, orderEvent) 把订单事件发到交换机 - 交换机按路由键投递到 order.queue 队列 - 发布者收到 Exchange 的确认回调publisher confirm接口就返回“预约成功”。与此同时系统里有三个消费者在监听这个队列的副本其实是三个不同队列绑定到同一交换机通知服务消费者消费消息调用短信 SDK 给用户和厨师发通知。派单服务消费者消费消息按地理位置和评分规则匹配最近的可服务厨师。日志服务消费者消费消息把关键事件写入审计日志表。这条链路的精妙之处在于短信服务临时挂掉不会影响主流程消息会留在队列里等恢复后继续消费派单逻辑变更也只需要改派单服务的代码其他服务完全不受影响。2. 环境搭建与 Spring Boot 集成配置2.1 RabbitMQ 的安装部署Windows 本地与 Docker 两条路安装这块我两种方案都试过先说结论个人开发机建议直接 Docker公司内网离线环境才需要考虑 Windows 安装包。如果你是在 Windows 上本地开发又不想装 Docker那就走 Erlang RabbitMQ 安装包的方式。Windows 方式先装 Erlang注意版本对应关系——RabbitMQ 3.9.x 对应 Erlang 23.2 以上3.12.x 需要 Erlang 25 以上版本对不上启动必报错再装 RabbitMQ Server MSI 安装包。装完用命令行进入 RabbitMQ 安装目录的 sbin 文件夹执行 rabbitmq-plugins enable rabbitmq_management 开启管理控制台插件然后浏览器访问 http://localhost:15672 就能看到管理界面默认账号密码是 guest/guest。注意guest 账号默认只能在 localhost 访问如果生产环境要远程管理必须新建一个自定义账号并赋予权限。Docker 方式就省心很多。我用的是 docker-compose直接贴配置version: 3.8 services: rabbitmq: image: rabbitmq:3.12-management container_name: rabbitmq ports: - 5672:5672 - 15672:15672 environment: RABBITMQ_DEFAULT_USER: admin RABBITMQ_DEFAULT_PASS: admin123 RABBITMQ_DEFAULT_VHOST: /myapp volumes: - rabbitmq_data:/var/lib/rabbitmq restart: unless-stopped volumes: rabbitmq_data:5672 是 AMQP 协议端口业务代码连接消息队列用的15672 是管理控制台 HTTP 端口。执行 docker-compose up -d 就能起一套带管理界面的 RabbitMQ管理地址是 http://localhost:15672账号 admin/admin123。在互联网上拉取镜像时一定要注意到 mirror 配置的问题你可以用配置镜像加速器的方式来解决不同云厂商加速器的配置方式不太一样自查一下即可。2.2 Spring Boot 版本与依赖的对应关系Spring Boot 从 2.x 到 3.x 的演进过程中spring-boot-starter-amqp 的协调版本一直在变不同 Spring Boot 版本对 AMQP 客户端的默认支持也不同。这里我整理了一份实际工作中验证过的版本对应关系Spring Boot 版本spring-boot-starter-amqp 默认 AMQP 客户端版本备注2.1.x2.1.x老项目仍在用需要手动处理 Jackson 兼容2.3.x2.2.x一般推荐 2.3 以上2.6.x2.4.x要注意 Springfox 3.0.0 集成时的路径匹配策略问题2.7.x2.4.x最后一个 2.x 稳定版本生产环境选型较多3.0.x 及以上3.0.x必须要 JDK 17网络上也常被检索到如果你搜到的是 Spring Boot 2.1 集成的示例放到 2.6 项目里大概率会遇到两个问题一个是 springfox 3.0.0Swagger 相关在 Spring Boot 2.6 以上会因为路径匹配策略从 AntPathMatcher 改为 PathPatternParser 而启动报错解决办法是在 application.yml 里加一行配置spring: mvc: pathmatch: matching-strategy: ant_path_matcher另一个是 RabbitMQ 连接工厂的配置项在 2.x 版本之间有少量术语调整。整体来说Spring Boot 2.6 和 2.7 是现在最常用的稳定版本下面代码示例我都以 Spring Boot 2.7 JDK 8/11 为准如果你用的是 3.x差异我会在文中标注。2.3 Spring Boot 集成 RabbitMQ 的基础配置在 pom.xml 引入依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency然后在 application.yml 里做基础配置。我给的配置比网上大多数教程要多因为我把生产者确认、消费者手动 ACK、消息重试这些生产环境必须的配置都写进去了spring: rabbitmq: host: localhost port: 5672 username: admin password: admin123 virtual-host: /myapp # 生产者端开启发送确认回调 publisher-confirm-type: correlated # 生产者端开启消息投递失败回退 publisher-returns: true template: mandatory: true listener: simple: # 消费者手动 ACK acknowledge-mode: manual # 初始并发消费者数 concurrency: 5 # 最大并发消费者数 max-concurrency: 15 # 每次预取消息数量 prefetch: 20 retry: enabled: true max-attempts: 3 initial-interval: 2000 multiplier: 2.0这几个参数每个都值得说清楚publisher-confirm-type: correlated 表示生产者发送的消息只要被交换机接收就会回调 ConfirmCallback告诉你消息有没有送达到交换机。这是防丢消息的第一道保险。template.mandatory: true 配合 publisher-returns 使用当消息不能路由到任何队列时会触发 ReturnedMessage 回调避免消息被静默丢弃。acknowledge-mode: manual 是消费者手动确认。默认 auto 模式在消费者方法抛出异常时消息会不断重试甚至导致消息被重复消费手动模式让你可以精确控制“什么时候算处理成功”。prefetch 是消费者单次从队列预取的消息数量。设得太大容易导致消息在消费者本地堆积设得太小吞吐量低。一般经验值是并发数乘以 3~5我这边 5 个初始消费者配 20 的 prefetch 是实测比较稳的。3. 核心代码实现与关键细节解析3.1 交换机、队列和绑定的声明策略网上很多教程喜欢在配置类里用 Bean 把队列、交换机、绑定全部声明好我一开始也是这么写的后来踩了改路由规则的坑——每次改代码重启都会重新声明一次万一某次手误把队列参数改错可能直接报“inequivalent arg”错误。所以我的建议是开发/测试环境用代码声明。好处是项目克隆下来直接就能跑不依赖手动在管理控制台建队列。生产环境建议在管理控制台或通过启动脚本声明一次代码里只写消费者和生产者的名字不要让应用自动声明关键队列。我这里给出开发环境标准的配置类写法Configuration public class RabbitMQConfig { // 订单交换机 Bean public DirectExchange orderExchange() { // 第一个参数是交换机名称第二个是是否持久化第三个是是否自动删除 return new DirectExchange(order.exchange, true, false); } // 订单队列 Bean public Queue orderQueue() { // durabletrue 表示队列持久化消息持久化 队列持久化才能在 RabbitMQ 重启后不丢消息 return QueueBuilder.durable(order.queue).build(); } // 绑定关系order.exchange 通过路由键 order.created 将消息投递到 order.queue Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()) .to(orderExchange()) .with(order.created); } }这里有两个持久化要分清一个是 exchange 和 queue 的 durable 属性管的是组件本身的存亡另一个是消息发送时 MessageDeliveryMode 是否持久化管的是消息数据是否写盘。两样必须同时满足RabbitMQ 重启后才不丢消息。Spring Boot 默认 convertAndSend 发送的消息就是持久化的但如果你手动构造 MessageProperties 可能会忽略这一点要注意。3.2 生产者代码怎么发消息、怎么确认消息到了交换机生产者这边我封装了一个消息发送工具把 confirm 回调和 return 回调都暴露出来方便排查线上问题。这是完整可运行的核心逻辑Service public class OrderEventPublisher { Autowired private RabbitTemplate rabbitTemplate; PostConstruct public void init() { // 消息到达交换机但路由不到任何队列时触发 rabbitTemplate.setMandatory(true); rabbitTemplate.setReturnsCallback(returned - { log.error(消息被退回: exchange{}, routingKey{}, body{}, replyCode{}, replyText{}, returned.getExchange(), returned.getRoutingKey(), new String(returned.getMessage().getBody()), returned.getReplyCode(), returned.getReplyText()); }); // 消息发送确认回调acktrue 表示交换机已接收 rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - { if (ack) { log.info(消息发送成功: {}, correlationData.getId()); } else { log.error(消息发送失败: {}, cause: {}, correlationData.getId(), cause); } }); } public void publishOrderCreated(OrderEvent event) { CorrelationData correlationData new CorrelationData(event.getEventId()); rabbitTemplate.convertAndSend(order.exchange, order.created, event, correlationData); } }CorrelationData 里我放的是 eventId这个 ID 一般就是 UUID它有两个用途一是把 confirm 回调和某条具体消息关联起来方便日志排查二是在消费端做幂等去重时会用到同一个 ID。我一直建议这个 ID 必须由业务生成而不是让框架自动生成否则出了问题你连是哪笔订单的消息都不知道。3.3 消费者代码手动 ACK 的正确姿势消费者是整个链路里最容易出问题的环节。我见过太多人只写了一个 RabbitListener 方法体用默认的 AUTO 确认模式方法里一旦抛异常消息就会一直重试把日志打满不说还可能让消息堆在 unacked 状态。我的标准写法是手动 ACKComponent public class OrderConsumer { Autowired private OrderService orderService; Autowired private RedisTemplateString, String redisTemplate; RabbitListener(queues order.queue, concurrency 5-15) public void onOrderCreated(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); String messageId message.getMessageProperties().getMessageId(); // 第一步幂等检查 Boolean firstConsume redisTemplate.opsForValue() .setIfAbsent(order:consumed: messageId, 1, Duration.ofHours(1)); if (firstConsume null || !firstConsume) { // 已经消费过确认并跳过 channel.basicAck(deliveryTag, false); return; } try { OrderEvent event JSON.parseObject(new String(message.getBody()), OrderEvent.class); // 真正业务逻辑“实现订单处理” orderService.handleOrderCreated(event); // 处理成功手动确认 channel.basicAck(deliveryTag, false); log.info(订单事件处理成功, orderId{}, event.getOrderId()); } catch (Exception e) { // 处理失败不确认也不重投进入死信队列或记录 log.error(订单事件处理失败, messageId{}, messageId, e); if (message.getMessageProperties().getRedelivered()) { // 已经重投过一次说明消费端始终有问题直接拒绝并让消息进入死信队列 channel.basicReject(deliveryTag, false); } else { // 首次失败requeuetrue 放回队列重试 channel.basicNack(deliveryTag, false, true); } } } }手动 ACK 的三个关键方法我把区别理一下basicAck确认消息处理成功消息从队列移除。basicNack否定消息第三个参数 multiple 表示是否批量否定requeue 表示是否重新放回队列。我这里首次失败 requeuetrue放回队列让别的消费者再次尝试。basicReject拒绝消息和 basicNack 的区别是不能批量处理。我设置 requeuefalse 是配合死信队列使用让“怎么都处理不了”的消息进入死信队列方便后面排查。为什么 rejection 前要判断 redelivered 标志如果不判断一个消费者每次都处理失败又每次 requeue消息就永远在队列和消费者之间打转形成死循环。判断 redelivered 就是为了限制无意义的重试轮次。当然你配置了 spring.rabbitmq.listener.simple.retry 之后重试逻辑由 Spring 管理不过手动 ACK 模式下最好还是自己在代码里控制重试策略更直观。3.4 JSON 消息序列化这是个隐藏大坑Spring Boot 的 RabbitTemplate 默认使用 Java 原生序列化JdkSerializationRedisSerializer 的思路类似但这里其实是 SimpleMessageConverter。默认 Sequence 化的结果是消息在 RabbitMQ 管理界面里是一堆看不懂的二进制而且跨语言消费方比如 C# 写的服务根本没法解析。所以一定要换成 JSON 序列化。我在配置类里加一个 BeanConfiguration public class RabbitMQMessageConverterConfig { Bean public MessageConverter messageConverter() { Jackson2JsonMessageConverter converter new Jackson2JsonMessageConverter(); // 把所有消息的 content-type 设为 application/json ObjectMapper objectMapper new ObjectMapper(); objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY); objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); converter.setObjectMapper(objectMapper); return converter; } }配置了 Jackson2JsonMessageConverter 之后生产者 convertAndSend 传对象时会自动序列化成 JSON消费者那边可以在 RabbitListener 方法入参直接写 OrderEvent 对象框架自动反序列化非常方便。注意 FAIL_ON_UNKNOWN_PROPERTIES 设置为 false这样即使生产者多加了字段老消费者也不会反序列化失败。还有个细节——header 里的 content-type。如果生产端发的消息 content-type 不是 application/json而消费端配置了 Jackson 转换器消费端可能直接报“无法转换”。排查时看管理控制台的 Message properties 就能定位。4. 常见问题与排查技巧实录4.1 启动失败连接被拒与 Erlang 版本不匹配这个问题出现频率最高尤其是 Windows 本地环境。现象是 Spring Boot 启动时报 Caused by: java.net.ConnectException: Connection refused: connect。排查思路按顺序走检查 RabbitMQ 进程是否在跑。在 Windows 上可以通过服务管理services.msc查看 RabbitMQ 服务状态或者在命令行执行 rabbitmqctl status 看是否输出正常运行信息。检查端口。telnet localhost 5672 能不能通不通就检查防火墙。Windows 安装场景重点检查 Erlang 版本。RabbitMQ 3.11 以上要求 Erlang 25版本不匹配时服务能启动但端口拒绝连接很迷惑。解决方式就是把 Erlang 卸了装上 RabbitMQ 对应要求的大版本号。Docker 场景检查端口映射是否写反了。注意是宿主机端口在前、容器端口在后。检查账号权限。控制台里确认这个账号有没有配置对应 vhost 的资源权限默认账号 guest 只在 localhost 生效。4.2 消息在队列里堆积消费者没人消费我遇到过一次比较极端的情况管理控制台里队列的 Ready 数量涨到了几十万但是没有任何消费者连接。先不是急着调并发而是排查消费者是否真正启动。按照以往的经验这种情况以下几个原因最常见RabbitListener 所在类没有被 Spring 扫描到方法根本没注册。消费者方法入参和实际消息类型不匹配反序列化报错但异常被吞了。这种情况在日志里能看到明显报错比如 ClassCastException。并发数配了 0 或者 prefetch 配了 0。concurrency 必须大于等于 1prefetch 建议不小于 1不然消费者就变成了“只看不取”。消费者处理消息太快但业务处理时间过长导致 unacked 数量很大但这不是没消费是消费速度跟不上。如果确认代码没问题可以从 RabbitMQ 的 Connections 和 Channels 页面看消费者是否在线。注意 Channel 里有一个 Prefetch 字段能看到实际生效的预取数量这个和你的配置可能不同以管理界面显示的为准。4.3 消息丢失从生产者到消费者的三层保护“消息发了但消费者没收到”是面试题里最高频的考点也是实战里最可怕的故障。消息从生产到消费要经过三段每一段都要保护第一段生产者 - 交换机。开启 publisher-confirm-type: correlated如果交换机没收到消息ConfirmCallback 里 ack 会是 false。注意如果消息连交换机都没到那一定是网络问题或连接配置问题检查连接配置中的 host/port/vhost。第二段交换机 - 队列。开启 mandatory returns 回调。如果交换机根据路由键找不到队列消息会被退回生产者。这里有个反直觉的点交换机本身收到消息且回调 ack 是 true但消息最终没进队列这不算“发送失败”所以在 returns 回调里一定要记录日志告警。第三段消费者处理。队列持久化 消息持久化 手动 ACK缺一不可。RabbitMQ 重启后持久化的队列和消息会恢复手动 ACK 确保消费者没处理完时消息不会从队列删除。三层都堵上了消息丢失的概率才能降到足够低。网上很多教程只讲了 queue durable 和 message persistent忽略生产者确认是远远不够的。4.4 重复消费与幂等处理重复消费是消息队列绕不开的课题哪怕你确认机制再完善消费者在 ACK 之后宕机、消息重投或者生产端因为网络抖动重发都会导致同一条消息被消费两次。业务上必须做幂等。我的做法是每条消息带一个全局唯一 messageId一般就是业务主键事件类型UUID 的组合消费者处理前先查 Redis没处理过继续执行处理成功后把 messageId 写入 Redis设置过期时间一般 24 小时。处理过直接 ACK 丢弃不重复执行业务。这里有一个容易忽略的细节Redis 写入一定要发生在业务执行之后而不是之前。如果先写 Redis 再执行业务万一业务执行失败但 Redis 里已经有了标记这条消息就永远不会被重试了。顺序应该是“查标记 - 执行业务 - 写标记 - ACK”这样即使业务失败消息还能重投下个消费者可以再次尝试。如果你用的是数据库可以建一张消息消费记录表用消息 ID 建唯一索引插入成功的消费者才继续执行效果一样。4.5 消费者处理慢导致 SQL 超时的问题这是一个比较隐蔽的生产问题队列消费逻辑里如果调用了数据库而 SQL 执行超过数据库驱动或连接池的最大等待时间连接会抛异常进而导致消费者线程处理失败、消息不断 requeue最终形成“消息积压 数据库连接池耗尽”的恶性循环。针对“JVM 或者 Spring Boot 会设置一个 SQL 执行 10 秒自动关闭吗”这类疑问要说明一下Spring Boot 本身并不会强制“SQL 执行 10 秒自动关闭”真正起作用的是连接池配置。比如 HikariCP 的 connection-timeout默认 30000ms和 socketTimeout数据库驱动自己的 socketTimeout以及 MySQL 的 wait_timeout。如果一条 SQL 执行超过 10 秒大概率是查询计划有问题或锁等待应优先优化 SQL而不是盲目调大超时。处理建议在消费者执行的 SQL 上设置合理的超时时间比如 MySQL JDBC 连接串加 socketTimeout5000避免一条慢 SQL 拖住整个消费线程同时给消费线程池配置独立的连接池如果量大的话不要和 Web 请求共用连接池防止互相干扰。另外项目里像“Spring Boot MyBatis 实现数据库字段级加密”这种场景字段加密后做等值查询会失效因为数据库里存的是密文。这类查询不能直接 SQL 匹配通常需要在应用层解密后根据业务规则处理或使用加密算法比如确定性加密来支持等值查询。这事在消费者里处理时要格外注意如果查询条件依赖加密字段要确保解密逻辑放在消费者线程里也能拿到正确的密钥上下文否则可能查出空数据业务静默失败消息被重复消费时也发现不了问题。4.6 Spring Boot 2.6 与 Springfox 3.0.0 的版本冲突我在一个 Spring Boot 2.6.6 的项目里集成 RabbitMQ 和 Swagger 时启动直接报错 Failed to start bean documentationPluginsBootstrapper; nested exception is java.lang.NullPointerException。原因是 Spring Boot 2.6 默认的 Spring MVC 路径匹配从 AntPathMatcher 变成了 PathPatternParser而 Springfox 3.0.0 还没有适配这种模式。解决办法就是在 application.yml 加上我前面提到的配置spring: mvc: pathmatch: matching-strategy: ant_path_matcher这个问题本身跟 RabbitMQ 没直接关系但项目一大了各种组件版本叠加这种问题很常见。排查思路就是看启动日志里最先出现的异常堆栈顺藤摸瓜找到是哪个框架和 Spring Boot 版本不兼容。集成消息队列时一定也要注意 starter 版本与 Boot 版本的对应关系。4.7 C# 或者其他语言客户端消费 JSON 消息的兼容性问题“C# 使用 RabbitMQ 推送”这类的场景很常见——Java 服务发消息C# 服务收消息或者反过来。如果 Java 端用的是默认 JDK 序列化C# 那边能收到一堆 bytes 但根本解析不了。这个问题我在前面 JSON 序列化部分已经埋了伏笔这里再强调一次跨语言场景必须统一用 JSON 消息格式。具体做法Java 端配置 Jackson2JsonMessageConverter确保发送时 content-type 是 application/json。C# 端用 RabbitMQ.Client 消费时把 ReadOnlyMemory 转成 UTF-8 字符串然后反序列化成自己的 Model。程序里只需要知道字段名不依赖 Java 类。不要在 JSON 里放 Java 特有类型如 LocalDateTime 序列化后的复杂格式建议统一为字符串的时间格式yyyy-MM-dd HH:mm:ss。这样 C# 端解析时直接 Parse 字符串简单可靠。4.8 官方管理控制台查看消息积压与系统状态排查 RabbitMQ 问题时管理控制台是最高效的入口。我来把常用查看项说明一下Overview 页签看全局的消息速率、连接数、通道数、节点状态。如果节点显示 red说明 Erlang 虚拟机或磁盘空间可能有问题。Connections 页签查看当前连接的生产者和消费者Connection 状态是 running 还是 blocked。blocked 说明可能触发了内存或磁盘告警默认 40MB 内存高水位、1GB 磁盘低水位到达后会阻塞生产者的写入。Queues 页签看每个队列的 Ready待消费和 Unacked已发送未确认数量。如果 Unacked 持续很高说明消费者处理速度跟不上如果 Ready 不断上涨说明生产速度远大于消费处理速度。两个数字长期不归零就要优化消费者性能了。消息的一个小技巧在 Queues 页签点进队列有一个 Get messages 功能可以手动拉取一条消息看看 body 内容帮助确认消息序列化格式是否是 JSON。4.9 死信队列与延迟队列的实现死信队列DLX在生产环境的地位相当于保险丝。业务处理失败且确认无法恢复时把消息扔进死信队列由独立消费者记录下来后续人工处理或补偿而不是在同一个队列里无限重试。延迟队列的实现原理其实是用死信队列 TTL 实现的特殊用法。先看代码Configuration public class DelayQueueConfig { // 延迟队列消息先停在这里等 TTL 过期 Bean public Queue delayQueue() { return QueueBuilder.durable(order.delay.queue) .withArgument(x-dead-letter-exchange, order.exchange) .withArgument(x-dead-letter-routing-key, order.timeout) .build(); } // 真正消费延迟消息的队列 Bean public Queue timeoutQueue() { return QueueBuilder.durable(order.timeout.queue).build(); } // 死信交换机 Bean public DirectExchange orderExchange() { return new DirectExchange(order.exchange, true, false); } // 绑定延迟队列到交换机路由键 order.delay Bean public Binding delayBinding() { return BindingBuilder.bind(delayQueue()) .to(orderExchange()) .with(order.delay); } // 绑定真正消费队列到交换机路由键 order.timeout Bean public Binding timeoutBinding() { return BindingBuilder.bind(timeoutQueue()) .to(orderExchange()) .with(order.timeout); } }发送延迟消息时设置消息的 TTL 为 30 分钟MessagePostProcessor processor message - { message.getMessageProperties().setExpiration(1800000); return message; }; rabbitTemplate.convertAndSend(order.exchange, order.delay, event, processor);原理是这样的消息带着 TTL 先进入 order.delay.queue30 分钟没人消费它就过期了。过期消息不会被自动删除而是根据队列里死信交换机的配置被投递到 order.exchange然后通过 order.timeout 路由键进入 order.timeout.queue。此时 RabbitListener(queues order.timeout.queue) 的消费者拿到消息做超时未支付取消预约的处理。这个方案有一个很小的缺陷如果队列里有多条消息设置不同的 TTLRabbitMQ 只会在队头消息过期时才检查后续消息不满足严格“按 TTL 从小到大投递”的需求。对大多数业务场景够用了。如果真要秒级精确的延迟任务建议评估 RocketMQ 的定时消息或 Redis 的 ZSet 方案。5. 高级话题与运维侧经验5.1 消费性能调优全链路参数配合记住一个公式队列吞吐上限 ≈ 并发消费者数 × 每条消息处理时间。想要提升消费能力就从这个公式找突破点。Spring 里并发消费者数由 concurrency 和 max-concurrency 控制。它们的关系是这样初始启动 5 个消费者如果队列里积压消息持续超过一定阈值RabbitMQ 会动态扩容到 15 个如果队列空了又会缩回 5 个。这个机制挺实用但要注意不能依赖无限扩容因为每个消费者都占用一个 TCP 连接和 Channel消费者太多会争抢 CPU 和数据库连接。prefetch 参数需要结合业务耗时来调。如果业务处理很快毫秒级prefetch 调大能提高吞吐如果业务处理很慢例如调外部 APIprefetch 不宜过大否则消息都积压在 unacked 状态超出 broker 的内存限制反而触发告警。另外一点消费者方法里尽量只做轻量操作把耗时操作交给单独的线程池去执行。如果消费者里还需要调用外部 API 或数据库建议给它们设置合理的超时时间。之前遇到一个坑RabbitListener 线程池默认大小不够外面任务一多线程被占满导致消费者处理停滞。这种情况下可以自定义 RabbitListenerContainerFactory 并配置它的线程池参数。5.2 消息幂等性设计模式除了 Redis 去重实际上还有几种幂等方案根据项目技术栈选择数据库唯一约束在消费记录表建立消息 ID 的唯一索引重复插入会报错利用这个特性实现幂等。适合没有 Redis、数据强一致要求高的项目。乐观锁业务表加 version 字段消费时先查再比对版本更新时带上 version 条件SQL 影响行数为 0 说明已经处理过了。状态机约束订单状态流转例如从“待付款”到“已关闭”是单向的如果消费到已关闭订单的消息直接丢弃。业务本身具有幂等语义时最简单。我特别推荐把“消费记录表”或 Redis 键的命名规范做成业务类型 业务主键 事件类型比如 order:paid:20240312104512345。这样排查问题时按订单号就能捞出一整条事件链。5.3 消息监控与告警RabbitMQ 的 HTTP API 提供了队列积压量的查询接口可以结合 Prometheus Grafana 或自研定时任务做监控。我常用的是直接调用管理 APIcurl -u admin:admin123 http://localhost:15672/api/queues/myapp/order.queue | jq .messages_ready重点监控指标其实就这几个messages_ready待消费数、messages_unacknowledged未确认数、consumers消费者数量。如果 messages_ready 持续上涨检查消费者如果 messages_unacknowledged 飙升检查消费者处理时间是否出现异常。告警阈值配置建议积压超过 1000 条或 unacked 持续 5 分钟超过 100 时通知。这个数字要根据业务量调整量级大的系统阈值可以放宽些。5.4 高可用部署单机 RabbitMQ 不适合做生产环境的主方案出现磁盘满了、内存溢出或节点宕机整个消息链路就断了。生产环境推荐镜像队列模式至少三节点集群队列在每个节点都有副本一个节点挂了其他节点还能继续服务。搭建三节点集群用 Docker Compose 或者 Kubernetes Operator 都可以。核心配置是让节点发现彼此然后通过管理控制台把队列设置为镜像模式或使用 Quorum Queue。Quorum Queue 是 RabbitMQ 3.8 之后推荐的高可用队列类型它基于 Raft 协议比经典镜像队列更健壮推荐新项目直接用 Quorum Queue。不过 Quorum Queue 对消息顺序的保证和经典队列略有差异使用前先看官方文档确认是否符合你的应用场景。在代码层面Spring Boot 连接多节点时配置 addressList 而不是单个 hostspring: rabbitmq: addresses: node1:5672,node2:5672,node3:5672 username: admin password: admin123这样的话客户端感知到节点故障就能自动切换到存活节点。客户端连接断开时Spring Boot 会按默认的重连策略恢复连接所以不需要自己写重连代码但连接恢复期间发送的消息会失败生产者端 ConfirmCallback 会收到 ackfalse这时候要考虑是否将失败消息暂存到本地表、后置任务重推。6. 真实业务场景复盘从零到一上线预约服务这部分我想把前面所有内容串起来用一个我实际参与过的上门烹饪预约服务消息队列部分做复盘方便你看完直接迁移到自己的“预约类”项目里。6.1 需求场景拆解“上门烹饪预约服务系统”的核心流程用户在小程序选择厨师、预约时间、菜品提交预约订单。系统自动为订单分配一个厨师。给用户发送预约成功的短信和公众号模板消息。厨师端 App 收到新的预约单推送。如果预约时间前 24 小时用户未取消系统生成备菜清单并提醒厨师。如果用户取消订单已派单的厨师端也要收到取消通知。如果所有动作都在下单请求里同步执行高峰期节假日午市会直接把 MySQL 和短信服务打崩。用 RabbitMQ 改造后的设计如下exchange: order.business.exchangeTopic 类型下单消息routing key order.created取消消息routing key order.cancelled延迟消息routing key order.pre.remind队列分配order.dispatch.queue - 绑定 order.created负责派单。order.notify.queue - 绑定 order.*负责短信/消息推送。order.log.queue - 绑定 order.*负责审计日志。order.remind.queue - 这是延迟队列消息先投到 delay 路由TTL 到期后进入 order.pre.remind负责提前 24 小时提醒。6.2 上线前压测数据与实践效果上线前我在测试环境用 Apache JMeter 做了压测500 并发同时下单模拟数据如下同一台 8C16G 机器方案下单接口 P95 耗时数据库连接池平均占用是否出现失败同步调用无 MQ3850ms28/30出现 5% 超时错误引入 RabbitMQ 异步320ms11/300 失败引入 MQ 后接口速度提升是立竿见影的。但也有代价——下单后的派单通知不是实时到达的从消息投递到消费者处理完成测试环境平均延迟在 50ms 以内用户体感无差别。6.3 这次项目踩到的三个印象深刻的坑第一个坑是消息被无限重试导致数据库连接池被占满。因为消费者方法里查 MySQL 超时抛了异常但错误被 try-catch 吞掉没有向上抛Spring Boot 不知道处理失败于是这条消息一直在“已投递”状态但业务实际没成功。后来我把所有消费者方法都改成“异常必须向上抛或者手动 NACK”避免静默失败。这个经验很重要消息队列里处理消息绝对不能把异常吞了不回 ACK 也不 NACK。第二个坑是 Redis 幂等标记设置过早。本来想用 Redis setIfAbsent 做幂等但当时把 setIfAbsent 放在业务处理之前了结果业务处理失败后消息重试时幂等标记已经存在导致这条消息永远无法被真正处理。后来改成“Redis 标记成功”放到业务成功之后才彻底解决。第三个坑是生产环境 RabbitMQ 管理控制台账号被误操作删除了。当时我用的是 guest 账号远程登录RabbitMQ 默认禁止 guest 远程访问导致连接超时。后来规范起来所有环境统一创建专用账号并严格按照最小权限分配 vhost。这个习惯一直保留到现在。6.4 面试里反复被问到的 RabbitMQ 问题既然热搜词里也包含“RabbitMQ 面试题”我顺手把高频面试题的核心答案整理出来方便跳槽的人复习也方便面试官快速建立对候选人的判断框架如何保证消息不丢失生产者 confirm 队列/消息持久化 消费者手动 ACK 合理配置死信队列。如何保证消息不重复消费消费方幂等常用 Redis setIfAbsent 或数据库唯一索引。如何保证消息顺序消费一个队列只对应一个消费者或者用 routing key 将同一订单的消息路由到同一队列并开启单消费者在消费者内部串行处理。RabbitMQ 本身不保证全局有序但可以保证单队列内有序。堆积了几百万条消息怎么办先查消费者是否在线、是否有异常临时紧急处理可以有几种典型做法扩展消费者并发、加机器修复消费者让消费速度提升紧急情况下甚至可以写脚本直接从队列里把消息转存到数据库再慢慢回放。延迟队列怎么实现TTL 死信交换机具体可参考前面代码。7. 一些个人的实操心得项目用到第三年我开始意识到 RabbitMQ 最大的坑其实不在 RabbitMQ 本身而在消息语义的设计上。如果你的团队还没有统一消息事件的定义格式、日志链路追踪方式消息一多起来就会很痛苦。我从项目里沉淀下来几条个人的心得分享给正在使用或准备使用 RabbitMQ 的朋友第一所有消息必须有统一的事件 ID并且贯穿始终。生产者的 CorrelationData、消费者 Redis 幂等键、日志链路里的 traceId都建议使用同一个 ID。这样一条消息从投递到消费到业务落库全程可以串联查询。第二队列和交换机尽量按照业务模块命名并且把 routing key 的设计当成 API 设计一样认真。比如“预约服务”相关的路由键建议统一格式为 order.created、order.cancelled、order.reminder。从路由键的名称就能看出业务语义避免时间长了之后连写这段代码的人都忘了某个 key 代表什么意思。第三消费者代码里尽量加上注解 RabbitListener 的 concurrency 参数时把这个参数放在常量类里统一管理不要散落到各个方法上。同一套环境不同消费者的并发数应该尽量一致避免某个消费者并发过高把数据库打满。第四存储消息内容和业务数据的数据库尽量分开。不要让消费者直接把队列消费和主要的订单库写操作混在一起否则消费高峰期的数据库负载会影响 Web 请求的稳定性。如果团队资源有限不准备单独建库至少要把消费者处理核心逻辑放到独立的 Service 里和 Controller 层读写保持一定的隔离。写到这里关于 Spring Boot 集成 RabbitMQ 的主要技术点和实战经验基本都覆盖了。最后再分享一个我最近实践的小技巧如果你的项目里同时有多个 RabbitListener 监听多个队列但不同队列的优先级不同可以定制多个 RabbitListenerContainerFactory给不同队列设置不同的 prefetch 和 concurrency 参数。比如订单队列并发设高一些通知队列并发设低一些这样资源分配更合理。不过这属于基线配置之后的调优手段刚刚接触 RabbitMQ 时还是从默认配置跑通再逐步调优比较稳妥。 Spring Boot集成RabbitMQ实战：架构设计、可靠性保障与踩坑记录
Spring Boot 集成 RabbitMQ 这事我从实际项目角度聊聊我的做法和踩过的坑。很多人在预约服务系统、订单通知、异步任务这些场景里选型消息队列最终都落在 RabbitMQ 上——因为它轻量、可靠、路由灵活而且 Spring Boot 对它有非常完善的自动配置。这篇文章我会从整体设计讲起把环境搭建、核心代码、可靠性保障、常见问题排查全部过一遍适合刚准备把 RabbitMQ 用进 Spring Boot 项目的人也适合已经用起来但被重复消费、消息丢失、积压这些问题折腾过的人。1. 项目整体设计与核心思路1.1 为什么是 RabbitMQ 而不是别的消息队列我在不少项目里做过消息中间件选型如果团队技术栈是 Java Spring BootRabbitMQ 往往是最稳的选择。它能解决的核心问题就三个异步、解耦、削峰。拿热搜里那个《基于 Spring Boot 的上门烹饪预约服务系统》来说用户下单后要去通知厨师、发短信、记录日志如果这些全在请求线程里同步做接口响应时间很容易从 200ms 冲到 2s。引入 RabbitMQ 之后下单接口只需要把订单事件丢进队列马上返回“下单成功”后面所有耗时的动作都交给消费者异步去做。这就是异步带来的直接收益。那为什么不选 KafkaKafka 的消息模型是分区追加日志吞吐量极高适合大数据量、日志采集、流处理场景。但它部署依赖 ZooKeeper新版本已经有 KRaft 模式运维成本更高而且 RabbitMQ 在复杂路由规则上的表现更好。在一个预约服务这类中小规模业务里每秒几百上千条消息就顶天了RabbitMQ 完全扛得住没必要引入 Kafka 的运维复杂度。RabbitMQ 另一个很典型的应用是延迟队列比如“用户下单后 30 分钟未支付自动取消”“预约时间临近提醒厨师备菜”这类定时任务需求用死信交换机实现非常优雅不需要自己写 Quartz 轮询扫表。1.2 RabbitMQ 的几个核心概念务必在动手前搞清我见过不少新手在写代码前没搞清楚这些概念后面配置交换机、队列时一头雾水。确实需要先弄明白这几个核心对象生产者Producer发消息的一方在 Spring Boot 里就是注入 RabbitTemplate 的 Service 或 Controller。消费者Consumer接收消息的一方在 Spring Boot 里就是标注了 RabbitListener 的方法。交换机Exchange消息分发的中转站它自己不存消息只负责把带路由键的消息投递到匹配的队列。队列Queue真正存储消息的地方。绑定Binding把交换机和队列关联起来的规则相当于一条定义“什么路由键进什么队列”的连线。虚拟主机Virtual Host可以理解为一个独立的小型 RabbitMQ 实例不同项目之间通过 vhost 隔离。交换机有四种类型直接型Direct、主题型Topic、广播型Fanout、头部型Headers。实际项目里最常用的是 Direct 和 TopicFanout 用于广播场景Headers 基本用不到。我用生活化的类比来解释一下交换机就像一个快递分拨中心队列是各个片区的快递站点。你寄快递时填写的目的地地址就是路由键Routing Key分拨中心根据这个地址把包裹送到对应片区的站点。如果你是高级会员还可以指定“一定要走航空件”这就是绑定规则里的参数。这么一理解Exchange 和 Queue 的关系就清楚了队列负责存交换机负责转绑定规则负责告诉交换机怎么转。1.3 消息队列在预约业务中的典型调用链路实操之前我先画一条典型的业务链路出来。不要依赖 mermaid 图我用文字描述你跟着走一遍就明白了用户在小程序端发起预约 - Spring Boot 的 OrderService 保存订单状态为“待确认” - 同时调用 rabbitTemplate.convertAndSend(order.exchange, order.created, orderEvent) 把订单事件发到交换机 - 交换机按路由键投递到 order.queue 队列 - 发布者收到 Exchange 的确认回调publisher confirm接口就返回“预约成功”。与此同时系统里有三个消费者在监听这个队列的副本其实是三个不同队列绑定到同一交换机通知服务消费者消费消息调用短信 SDK 给用户和厨师发通知。派单服务消费者消费消息按地理位置和评分规则匹配最近的可服务厨师。日志服务消费者消费消息把关键事件写入审计日志表。这条链路的精妙之处在于短信服务临时挂掉不会影响主流程消息会留在队列里等恢复后继续消费派单逻辑变更也只需要改派单服务的代码其他服务完全不受影响。2. 环境搭建与 Spring Boot 集成配置2.1 RabbitMQ 的安装部署Windows 本地与 Docker 两条路安装这块我两种方案都试过先说结论个人开发机建议直接 Docker公司内网离线环境才需要考虑 Windows 安装包。如果你是在 Windows 上本地开发又不想装 Docker那就走 Erlang RabbitMQ 安装包的方式。Windows 方式先装 Erlang注意版本对应关系——RabbitMQ 3.9.x 对应 Erlang 23.2 以上3.12.x 需要 Erlang 25 以上版本对不上启动必报错再装 RabbitMQ Server MSI 安装包。装完用命令行进入 RabbitMQ 安装目录的 sbin 文件夹执行 rabbitmq-plugins enable rabbitmq_management 开启管理控制台插件然后浏览器访问 http://localhost:15672 就能看到管理界面默认账号密码是 guest/guest。注意guest 账号默认只能在 localhost 访问如果生产环境要远程管理必须新建一个自定义账号并赋予权限。Docker 方式就省心很多。我用的是 docker-compose直接贴配置version: 3.8 services: rabbitmq: image: rabbitmq:3.12-management container_name: rabbitmq ports: - 5672:5672 - 15672:15672 environment: RABBITMQ_DEFAULT_USER: admin RABBITMQ_DEFAULT_PASS: admin123 RABBITMQ_DEFAULT_VHOST: /myapp volumes: - rabbitmq_data:/var/lib/rabbitmq restart: unless-stopped volumes: rabbitmq_data:5672 是 AMQP 协议端口业务代码连接消息队列用的15672 是管理控制台 HTTP 端口。执行 docker-compose up -d 就能起一套带管理界面的 RabbitMQ管理地址是 http://localhost:15672账号 admin/admin123。在互联网上拉取镜像时一定要注意到 mirror 配置的问题你可以用配置镜像加速器的方式来解决不同云厂商加速器的配置方式不太一样自查一下即可。2.2 Spring Boot 版本与依赖的对应关系Spring Boot 从 2.x 到 3.x 的演进过程中spring-boot-starter-amqp 的协调版本一直在变不同 Spring Boot 版本对 AMQP 客户端的默认支持也不同。这里我整理了一份实际工作中验证过的版本对应关系Spring Boot 版本spring-boot-starter-amqp 默认 AMQP 客户端版本备注2.1.x2.1.x老项目仍在用需要手动处理 Jackson 兼容2.3.x2.2.x一般推荐 2.3 以上2.6.x2.4.x要注意 Springfox 3.0.0 集成时的路径匹配策略问题2.7.x2.4.x最后一个 2.x 稳定版本生产环境选型较多3.0.x 及以上3.0.x必须要 JDK 17网络上也常被检索到如果你搜到的是 Spring Boot 2.1 集成的示例放到 2.6 项目里大概率会遇到两个问题一个是 springfox 3.0.0Swagger 相关在 Spring Boot 2.6 以上会因为路径匹配策略从 AntPathMatcher 改为 PathPatternParser 而启动报错解决办法是在 application.yml 里加一行配置spring: mvc: pathmatch: matching-strategy: ant_path_matcher另一个是 RabbitMQ 连接工厂的配置项在 2.x 版本之间有少量术语调整。整体来说Spring Boot 2.6 和 2.7 是现在最常用的稳定版本下面代码示例我都以 Spring Boot 2.7 JDK 8/11 为准如果你用的是 3.x差异我会在文中标注。2.3 Spring Boot 集成 RabbitMQ 的基础配置在 pom.xml 引入依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency然后在 application.yml 里做基础配置。我给的配置比网上大多数教程要多因为我把生产者确认、消费者手动 ACK、消息重试这些生产环境必须的配置都写进去了spring: rabbitmq: host: localhost port: 5672 username: admin password: admin123 virtual-host: /myapp # 生产者端开启发送确认回调 publisher-confirm-type: correlated # 生产者端开启消息投递失败回退 publisher-returns: true template: mandatory: true listener: simple: # 消费者手动 ACK acknowledge-mode: manual # 初始并发消费者数 concurrency: 5 # 最大并发消费者数 max-concurrency: 15 # 每次预取消息数量 prefetch: 20 retry: enabled: true max-attempts: 3 initial-interval: 2000 multiplier: 2.0这几个参数每个都值得说清楚publisher-confirm-type: correlated 表示生产者发送的消息只要被交换机接收就会回调 ConfirmCallback告诉你消息有没有送达到交换机。这是防丢消息的第一道保险。template.mandatory: true 配合 publisher-returns 使用当消息不能路由到任何队列时会触发 ReturnedMessage 回调避免消息被静默丢弃。acknowledge-mode: manual 是消费者手动确认。默认 auto 模式在消费者方法抛出异常时消息会不断重试甚至导致消息被重复消费手动模式让你可以精确控制“什么时候算处理成功”。prefetch 是消费者单次从队列预取的消息数量。设得太大容易导致消息在消费者本地堆积设得太小吞吐量低。一般经验值是并发数乘以 3~5我这边 5 个初始消费者配 20 的 prefetch 是实测比较稳的。3. 核心代码实现与关键细节解析3.1 交换机、队列和绑定的声明策略网上很多教程喜欢在配置类里用 Bean 把队列、交换机、绑定全部声明好我一开始也是这么写的后来踩了改路由规则的坑——每次改代码重启都会重新声明一次万一某次手误把队列参数改错可能直接报“inequivalent arg”错误。所以我的建议是开发/测试环境用代码声明。好处是项目克隆下来直接就能跑不依赖手动在管理控制台建队列。生产环境建议在管理控制台或通过启动脚本声明一次代码里只写消费者和生产者的名字不要让应用自动声明关键队列。我这里给出开发环境标准的配置类写法Configuration public class RabbitMQConfig { // 订单交换机 Bean public DirectExchange orderExchange() { // 第一个参数是交换机名称第二个是是否持久化第三个是是否自动删除 return new DirectExchange(order.exchange, true, false); } // 订单队列 Bean public Queue orderQueue() { // durabletrue 表示队列持久化消息持久化 队列持久化才能在 RabbitMQ 重启后不丢消息 return QueueBuilder.durable(order.queue).build(); } // 绑定关系order.exchange 通过路由键 order.created 将消息投递到 order.queue Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()) .to(orderExchange()) .with(order.created); } }这里有两个持久化要分清一个是 exchange 和 queue 的 durable 属性管的是组件本身的存亡另一个是消息发送时 MessageDeliveryMode 是否持久化管的是消息数据是否写盘。两样必须同时满足RabbitMQ 重启后才不丢消息。Spring Boot 默认 convertAndSend 发送的消息就是持久化的但如果你手动构造 MessageProperties 可能会忽略这一点要注意。3.2 生产者代码怎么发消息、怎么确认消息到了交换机生产者这边我封装了一个消息发送工具把 confirm 回调和 return 回调都暴露出来方便排查线上问题。这是完整可运行的核心逻辑Service public class OrderEventPublisher { Autowired private RabbitTemplate rabbitTemplate; PostConstruct public void init() { // 消息到达交换机但路由不到任何队列时触发 rabbitTemplate.setMandatory(true); rabbitTemplate.setReturnsCallback(returned - { log.error(消息被退回: exchange{}, routingKey{}, body{}, replyCode{}, replyText{}, returned.getExchange(), returned.getRoutingKey(), new String(returned.getMessage().getBody()), returned.getReplyCode(), returned.getReplyText()); }); // 消息发送确认回调acktrue 表示交换机已接收 rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - { if (ack) { log.info(消息发送成功: {}, correlationData.getId()); } else { log.error(消息发送失败: {}, cause: {}, correlationData.getId(), cause); } }); } public void publishOrderCreated(OrderEvent event) { CorrelationData correlationData new CorrelationData(event.getEventId()); rabbitTemplate.convertAndSend(order.exchange, order.created, event, correlationData); } }CorrelationData 里我放的是 eventId这个 ID 一般就是 UUID它有两个用途一是把 confirm 回调和某条具体消息关联起来方便日志排查二是在消费端做幂等去重时会用到同一个 ID。我一直建议这个 ID 必须由业务生成而不是让框架自动生成否则出了问题你连是哪笔订单的消息都不知道。3.3 消费者代码手动 ACK 的正确姿势消费者是整个链路里最容易出问题的环节。我见过太多人只写了一个 RabbitListener 方法体用默认的 AUTO 确认模式方法里一旦抛异常消息就会一直重试把日志打满不说还可能让消息堆在 unacked 状态。我的标准写法是手动 ACKComponent public class OrderConsumer { Autowired private OrderService orderService; Autowired private RedisTemplateString, String redisTemplate; RabbitListener(queues order.queue, concurrency 5-15) public void onOrderCreated(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); String messageId message.getMessageProperties().getMessageId(); // 第一步幂等检查 Boolean firstConsume redisTemplate.opsForValue() .setIfAbsent(order:consumed: messageId, 1, Duration.ofHours(1)); if (firstConsume null || !firstConsume) { // 已经消费过确认并跳过 channel.basicAck(deliveryTag, false); return; } try { OrderEvent event JSON.parseObject(new String(message.getBody()), OrderEvent.class); // 真正业务逻辑“实现订单处理” orderService.handleOrderCreated(event); // 处理成功手动确认 channel.basicAck(deliveryTag, false); log.info(订单事件处理成功, orderId{}, event.getOrderId()); } catch (Exception e) { // 处理失败不确认也不重投进入死信队列或记录 log.error(订单事件处理失败, messageId{}, messageId, e); if (message.getMessageProperties().getRedelivered()) { // 已经重投过一次说明消费端始终有问题直接拒绝并让消息进入死信队列 channel.basicReject(deliveryTag, false); } else { // 首次失败requeuetrue 放回队列重试 channel.basicNack(deliveryTag, false, true); } } } }手动 ACK 的三个关键方法我把区别理一下basicAck确认消息处理成功消息从队列移除。basicNack否定消息第三个参数 multiple 表示是否批量否定requeue 表示是否重新放回队列。我这里首次失败 requeuetrue放回队列让别的消费者再次尝试。basicReject拒绝消息和 basicNack 的区别是不能批量处理。我设置 requeuefalse 是配合死信队列使用让“怎么都处理不了”的消息进入死信队列方便后面排查。为什么 rejection 前要判断 redelivered 标志如果不判断一个消费者每次都处理失败又每次 requeue消息就永远在队列和消费者之间打转形成死循环。判断 redelivered 就是为了限制无意义的重试轮次。当然你配置了 spring.rabbitmq.listener.simple.retry 之后重试逻辑由 Spring 管理不过手动 ACK 模式下最好还是自己在代码里控制重试策略更直观。3.4 JSON 消息序列化这是个隐藏大坑Spring Boot 的 RabbitTemplate 默认使用 Java 原生序列化JdkSerializationRedisSerializer 的思路类似但这里其实是 SimpleMessageConverter。默认 Sequence 化的结果是消息在 RabbitMQ 管理界面里是一堆看不懂的二进制而且跨语言消费方比如 C# 写的服务根本没法解析。所以一定要换成 JSON 序列化。我在配置类里加一个 BeanConfiguration public class RabbitMQMessageConverterConfig { Bean public MessageConverter messageConverter() { Jackson2JsonMessageConverter converter new Jackson2JsonMessageConverter(); // 把所有消息的 content-type 设为 application/json ObjectMapper objectMapper new ObjectMapper(); objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY); objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); converter.setObjectMapper(objectMapper); return converter; } }配置了 Jackson2JsonMessageConverter 之后生产者 convertAndSend 传对象时会自动序列化成 JSON消费者那边可以在 RabbitListener 方法入参直接写 OrderEvent 对象框架自动反序列化非常方便。注意 FAIL_ON_UNKNOWN_PROPERTIES 设置为 false这样即使生产者多加了字段老消费者也不会反序列化失败。还有个细节——header 里的 content-type。如果生产端发的消息 content-type 不是 application/json而消费端配置了 Jackson 转换器消费端可能直接报“无法转换”。排查时看管理控制台的 Message properties 就能定位。4. 常见问题与排查技巧实录4.1 启动失败连接被拒与 Erlang 版本不匹配这个问题出现频率最高尤其是 Windows 本地环境。现象是 Spring Boot 启动时报 Caused by: java.net.ConnectException: Connection refused: connect。排查思路按顺序走检查 RabbitMQ 进程是否在跑。在 Windows 上可以通过服务管理services.msc查看 RabbitMQ 服务状态或者在命令行执行 rabbitmqctl status 看是否输出正常运行信息。检查端口。telnet localhost 5672 能不能通不通就检查防火墙。Windows 安装场景重点检查 Erlang 版本。RabbitMQ 3.11 以上要求 Erlang 25版本不匹配时服务能启动但端口拒绝连接很迷惑。解决方式就是把 Erlang 卸了装上 RabbitMQ 对应要求的大版本号。Docker 场景检查端口映射是否写反了。注意是宿主机端口在前、容器端口在后。检查账号权限。控制台里确认这个账号有没有配置对应 vhost 的资源权限默认账号 guest 只在 localhost 生效。4.2 消息在队列里堆积消费者没人消费我遇到过一次比较极端的情况管理控制台里队列的 Ready 数量涨到了几十万但是没有任何消费者连接。先不是急着调并发而是排查消费者是否真正启动。按照以往的经验这种情况以下几个原因最常见RabbitListener 所在类没有被 Spring 扫描到方法根本没注册。消费者方法入参和实际消息类型不匹配反序列化报错但异常被吞了。这种情况在日志里能看到明显报错比如 ClassCastException。并发数配了 0 或者 prefetch 配了 0。concurrency 必须大于等于 1prefetch 建议不小于 1不然消费者就变成了“只看不取”。消费者处理消息太快但业务处理时间过长导致 unacked 数量很大但这不是没消费是消费速度跟不上。如果确认代码没问题可以从 RabbitMQ 的 Connections 和 Channels 页面看消费者是否在线。注意 Channel 里有一个 Prefetch 字段能看到实际生效的预取数量这个和你的配置可能不同以管理界面显示的为准。4.3 消息丢失从生产者到消费者的三层保护“消息发了但消费者没收到”是面试题里最高频的考点也是实战里最可怕的故障。消息从生产到消费要经过三段每一段都要保护第一段生产者 - 交换机。开启 publisher-confirm-type: correlated如果交换机没收到消息ConfirmCallback 里 ack 会是 false。注意如果消息连交换机都没到那一定是网络问题或连接配置问题检查连接配置中的 host/port/vhost。第二段交换机 - 队列。开启 mandatory returns 回调。如果交换机根据路由键找不到队列消息会被退回生产者。这里有个反直觉的点交换机本身收到消息且回调 ack 是 true但消息最终没进队列这不算“发送失败”所以在 returns 回调里一定要记录日志告警。第三段消费者处理。队列持久化 消息持久化 手动 ACK缺一不可。RabbitMQ 重启后持久化的队列和消息会恢复手动 ACK 确保消费者没处理完时消息不会从队列删除。三层都堵上了消息丢失的概率才能降到足够低。网上很多教程只讲了 queue durable 和 message persistent忽略生产者确认是远远不够的。4.4 重复消费与幂等处理重复消费是消息队列绕不开的课题哪怕你确认机制再完善消费者在 ACK 之后宕机、消息重投或者生产端因为网络抖动重发都会导致同一条消息被消费两次。业务上必须做幂等。我的做法是每条消息带一个全局唯一 messageId一般就是业务主键事件类型UUID 的组合消费者处理前先查 Redis没处理过继续执行处理成功后把 messageId 写入 Redis设置过期时间一般 24 小时。处理过直接 ACK 丢弃不重复执行业务。这里有一个容易忽略的细节Redis 写入一定要发生在业务执行之后而不是之前。如果先写 Redis 再执行业务万一业务执行失败但 Redis 里已经有了标记这条消息就永远不会被重试了。顺序应该是“查标记 - 执行业务 - 写标记 - ACK”这样即使业务失败消息还能重投下个消费者可以再次尝试。如果你用的是数据库可以建一张消息消费记录表用消息 ID 建唯一索引插入成功的消费者才继续执行效果一样。4.5 消费者处理慢导致 SQL 超时的问题这是一个比较隐蔽的生产问题队列消费逻辑里如果调用了数据库而 SQL 执行超过数据库驱动或连接池的最大等待时间连接会抛异常进而导致消费者线程处理失败、消息不断 requeue最终形成“消息积压 数据库连接池耗尽”的恶性循环。针对“JVM 或者 Spring Boot 会设置一个 SQL 执行 10 秒自动关闭吗”这类疑问要说明一下Spring Boot 本身并不会强制“SQL 执行 10 秒自动关闭”真正起作用的是连接池配置。比如 HikariCP 的 connection-timeout默认 30000ms和 socketTimeout数据库驱动自己的 socketTimeout以及 MySQL 的 wait_timeout。如果一条 SQL 执行超过 10 秒大概率是查询计划有问题或锁等待应优先优化 SQL而不是盲目调大超时。处理建议在消费者执行的 SQL 上设置合理的超时时间比如 MySQL JDBC 连接串加 socketTimeout5000避免一条慢 SQL 拖住整个消费线程同时给消费线程池配置独立的连接池如果量大的话不要和 Web 请求共用连接池防止互相干扰。另外项目里像“Spring Boot MyBatis 实现数据库字段级加密”这种场景字段加密后做等值查询会失效因为数据库里存的是密文。这类查询不能直接 SQL 匹配通常需要在应用层解密后根据业务规则处理或使用加密算法比如确定性加密来支持等值查询。这事在消费者里处理时要格外注意如果查询条件依赖加密字段要确保解密逻辑放在消费者线程里也能拿到正确的密钥上下文否则可能查出空数据业务静默失败消息被重复消费时也发现不了问题。4.6 Spring Boot 2.6 与 Springfox 3.0.0 的版本冲突我在一个 Spring Boot 2.6.6 的项目里集成 RabbitMQ 和 Swagger 时启动直接报错 Failed to start bean documentationPluginsBootstrapper; nested exception is java.lang.NullPointerException。原因是 Spring Boot 2.6 默认的 Spring MVC 路径匹配从 AntPathMatcher 变成了 PathPatternParser而 Springfox 3.0.0 还没有适配这种模式。解决办法就是在 application.yml 加上我前面提到的配置spring: mvc: pathmatch: matching-strategy: ant_path_matcher这个问题本身跟 RabbitMQ 没直接关系但项目一大了各种组件版本叠加这种问题很常见。排查思路就是看启动日志里最先出现的异常堆栈顺藤摸瓜找到是哪个框架和 Spring Boot 版本不兼容。集成消息队列时一定也要注意 starter 版本与 Boot 版本的对应关系。4.7 C# 或者其他语言客户端消费 JSON 消息的兼容性问题“C# 使用 RabbitMQ 推送”这类的场景很常见——Java 服务发消息C# 服务收消息或者反过来。如果 Java 端用的是默认 JDK 序列化C# 那边能收到一堆 bytes 但根本解析不了。这个问题我在前面 JSON 序列化部分已经埋了伏笔这里再强调一次跨语言场景必须统一用 JSON 消息格式。具体做法Java 端配置 Jackson2JsonMessageConverter确保发送时 content-type 是 application/json。C# 端用 RabbitMQ.Client 消费时把 ReadOnlyMemory 转成 UTF-8 字符串然后反序列化成自己的 Model。程序里只需要知道字段名不依赖 Java 类。不要在 JSON 里放 Java 特有类型如 LocalDateTime 序列化后的复杂格式建议统一为字符串的时间格式yyyy-MM-dd HH:mm:ss。这样 C# 端解析时直接 Parse 字符串简单可靠。4.8 官方管理控制台查看消息积压与系统状态排查 RabbitMQ 问题时管理控制台是最高效的入口。我来把常用查看项说明一下Overview 页签看全局的消息速率、连接数、通道数、节点状态。如果节点显示 red说明 Erlang 虚拟机或磁盘空间可能有问题。Connections 页签查看当前连接的生产者和消费者Connection 状态是 running 还是 blocked。blocked 说明可能触发了内存或磁盘告警默认 40MB 内存高水位、1GB 磁盘低水位到达后会阻塞生产者的写入。Queues 页签看每个队列的 Ready待消费和 Unacked已发送未确认数量。如果 Unacked 持续很高说明消费者处理速度跟不上如果 Ready 不断上涨说明生产速度远大于消费处理速度。两个数字长期不归零就要优化消费者性能了。消息的一个小技巧在 Queues 页签点进队列有一个 Get messages 功能可以手动拉取一条消息看看 body 内容帮助确认消息序列化格式是否是 JSON。4.9 死信队列与延迟队列的实现死信队列DLX在生产环境的地位相当于保险丝。业务处理失败且确认无法恢复时把消息扔进死信队列由独立消费者记录下来后续人工处理或补偿而不是在同一个队列里无限重试。延迟队列的实现原理其实是用死信队列 TTL 实现的特殊用法。先看代码Configuration public class DelayQueueConfig { // 延迟队列消息先停在这里等 TTL 过期 Bean public Queue delayQueue() { return QueueBuilder.durable(order.delay.queue) .withArgument(x-dead-letter-exchange, order.exchange) .withArgument(x-dead-letter-routing-key, order.timeout) .build(); } // 真正消费延迟消息的队列 Bean public Queue timeoutQueue() { return QueueBuilder.durable(order.timeout.queue).build(); } // 死信交换机 Bean public DirectExchange orderExchange() { return new DirectExchange(order.exchange, true, false); } // 绑定延迟队列到交换机路由键 order.delay Bean public Binding delayBinding() { return BindingBuilder.bind(delayQueue()) .to(orderExchange()) .with(order.delay); } // 绑定真正消费队列到交换机路由键 order.timeout Bean public Binding timeoutBinding() { return BindingBuilder.bind(timeoutQueue()) .to(orderExchange()) .with(order.timeout); } }发送延迟消息时设置消息的 TTL 为 30 分钟MessagePostProcessor processor message - { message.getMessageProperties().setExpiration(1800000); return message; }; rabbitTemplate.convertAndSend(order.exchange, order.delay, event, processor);原理是这样的消息带着 TTL 先进入 order.delay.queue30 分钟没人消费它就过期了。过期消息不会被自动删除而是根据队列里死信交换机的配置被投递到 order.exchange然后通过 order.timeout 路由键进入 order.timeout.queue。此时 RabbitListener(queues order.timeout.queue) 的消费者拿到消息做超时未支付取消预约的处理。这个方案有一个很小的缺陷如果队列里有多条消息设置不同的 TTLRabbitMQ 只会在队头消息过期时才检查后续消息不满足严格“按 TTL 从小到大投递”的需求。对大多数业务场景够用了。如果真要秒级精确的延迟任务建议评估 RocketMQ 的定时消息或 Redis 的 ZSet 方案。5. 高级话题与运维侧经验5.1 消费性能调优全链路参数配合记住一个公式队列吞吐上限 ≈ 并发消费者数 × 每条消息处理时间。想要提升消费能力就从这个公式找突破点。Spring 里并发消费者数由 concurrency 和 max-concurrency 控制。它们的关系是这样初始启动 5 个消费者如果队列里积压消息持续超过一定阈值RabbitMQ 会动态扩容到 15 个如果队列空了又会缩回 5 个。这个机制挺实用但要注意不能依赖无限扩容因为每个消费者都占用一个 TCP 连接和 Channel消费者太多会争抢 CPU 和数据库连接。prefetch 参数需要结合业务耗时来调。如果业务处理很快毫秒级prefetch 调大能提高吞吐如果业务处理很慢例如调外部 APIprefetch 不宜过大否则消息都积压在 unacked 状态超出 broker 的内存限制反而触发告警。另外一点消费者方法里尽量只做轻量操作把耗时操作交给单独的线程池去执行。如果消费者里还需要调用外部 API 或数据库建议给它们设置合理的超时时间。之前遇到一个坑RabbitListener 线程池默认大小不够外面任务一多线程被占满导致消费者处理停滞。这种情况下可以自定义 RabbitListenerContainerFactory 并配置它的线程池参数。5.2 消息幂等性设计模式除了 Redis 去重实际上还有几种幂等方案根据项目技术栈选择数据库唯一约束在消费记录表建立消息 ID 的唯一索引重复插入会报错利用这个特性实现幂等。适合没有 Redis、数据强一致要求高的项目。乐观锁业务表加 version 字段消费时先查再比对版本更新时带上 version 条件SQL 影响行数为 0 说明已经处理过了。状态机约束订单状态流转例如从“待付款”到“已关闭”是单向的如果消费到已关闭订单的消息直接丢弃。业务本身具有幂等语义时最简单。我特别推荐把“消费记录表”或 Redis 键的命名规范做成业务类型 业务主键 事件类型比如 order:paid:20240312104512345。这样排查问题时按订单号就能捞出一整条事件链。5.3 消息监控与告警RabbitMQ 的 HTTP API 提供了队列积压量的查询接口可以结合 Prometheus Grafana 或自研定时任务做监控。我常用的是直接调用管理 APIcurl -u admin:admin123 http://localhost:15672/api/queues/myapp/order.queue | jq .messages_ready重点监控指标其实就这几个messages_ready待消费数、messages_unacknowledged未确认数、consumers消费者数量。如果 messages_ready 持续上涨检查消费者如果 messages_unacknowledged 飙升检查消费者处理时间是否出现异常。告警阈值配置建议积压超过 1000 条或 unacked 持续 5 分钟超过 100 时通知。这个数字要根据业务量调整量级大的系统阈值可以放宽些。5.4 高可用部署单机 RabbitMQ 不适合做生产环境的主方案出现磁盘满了、内存溢出或节点宕机整个消息链路就断了。生产环境推荐镜像队列模式至少三节点集群队列在每个节点都有副本一个节点挂了其他节点还能继续服务。搭建三节点集群用 Docker Compose 或者 Kubernetes Operator 都可以。核心配置是让节点发现彼此然后通过管理控制台把队列设置为镜像模式或使用 Quorum Queue。Quorum Queue 是 RabbitMQ 3.8 之后推荐的高可用队列类型它基于 Raft 协议比经典镜像队列更健壮推荐新项目直接用 Quorum Queue。不过 Quorum Queue 对消息顺序的保证和经典队列略有差异使用前先看官方文档确认是否符合你的应用场景。在代码层面Spring Boot 连接多节点时配置 addressList 而不是单个 hostspring: rabbitmq: addresses: node1:5672,node2:5672,node3:5672 username: admin password: admin123这样的话客户端感知到节点故障就能自动切换到存活节点。客户端连接断开时Spring Boot 会按默认的重连策略恢复连接所以不需要自己写重连代码但连接恢复期间发送的消息会失败生产者端 ConfirmCallback 会收到 ackfalse这时候要考虑是否将失败消息暂存到本地表、后置任务重推。6. 真实业务场景复盘从零到一上线预约服务这部分我想把前面所有内容串起来用一个我实际参与过的上门烹饪预约服务消息队列部分做复盘方便你看完直接迁移到自己的“预约类”项目里。6.1 需求场景拆解“上门烹饪预约服务系统”的核心流程用户在小程序选择厨师、预约时间、菜品提交预约订单。系统自动为订单分配一个厨师。给用户发送预约成功的短信和公众号模板消息。厨师端 App 收到新的预约单推送。如果预约时间前 24 小时用户未取消系统生成备菜清单并提醒厨师。如果用户取消订单已派单的厨师端也要收到取消通知。如果所有动作都在下单请求里同步执行高峰期节假日午市会直接把 MySQL 和短信服务打崩。用 RabbitMQ 改造后的设计如下exchange: order.business.exchangeTopic 类型下单消息routing key order.created取消消息routing key order.cancelled延迟消息routing key order.pre.remind队列分配order.dispatch.queue - 绑定 order.created负责派单。order.notify.queue - 绑定 order.*负责短信/消息推送。order.log.queue - 绑定 order.*负责审计日志。order.remind.queue - 这是延迟队列消息先投到 delay 路由TTL 到期后进入 order.pre.remind负责提前 24 小时提醒。6.2 上线前压测数据与实践效果上线前我在测试环境用 Apache JMeter 做了压测500 并发同时下单模拟数据如下同一台 8C16G 机器方案下单接口 P95 耗时数据库连接池平均占用是否出现失败同步调用无 MQ3850ms28/30出现 5% 超时错误引入 RabbitMQ 异步320ms11/300 失败引入 MQ 后接口速度提升是立竿见影的。但也有代价——下单后的派单通知不是实时到达的从消息投递到消费者处理完成测试环境平均延迟在 50ms 以内用户体感无差别。6.3 这次项目踩到的三个印象深刻的坑第一个坑是消息被无限重试导致数据库连接池被占满。因为消费者方法里查 MySQL 超时抛了异常但错误被 try-catch 吞掉没有向上抛Spring Boot 不知道处理失败于是这条消息一直在“已投递”状态但业务实际没成功。后来我把所有消费者方法都改成“异常必须向上抛或者手动 NACK”避免静默失败。这个经验很重要消息队列里处理消息绝对不能把异常吞了不回 ACK 也不 NACK。第二个坑是 Redis 幂等标记设置过早。本来想用 Redis setIfAbsent 做幂等但当时把 setIfAbsent 放在业务处理之前了结果业务处理失败后消息重试时幂等标记已经存在导致这条消息永远无法被真正处理。后来改成“Redis 标记成功”放到业务成功之后才彻底解决。第三个坑是生产环境 RabbitMQ 管理控制台账号被误操作删除了。当时我用的是 guest 账号远程登录RabbitMQ 默认禁止 guest 远程访问导致连接超时。后来规范起来所有环境统一创建专用账号并严格按照最小权限分配 vhost。这个习惯一直保留到现在。6.4 面试里反复被问到的 RabbitMQ 问题既然热搜词里也包含“RabbitMQ 面试题”我顺手把高频面试题的核心答案整理出来方便跳槽的人复习也方便面试官快速建立对候选人的判断框架如何保证消息不丢失生产者 confirm 队列/消息持久化 消费者手动 ACK 合理配置死信队列。如何保证消息不重复消费消费方幂等常用 Redis setIfAbsent 或数据库唯一索引。如何保证消息顺序消费一个队列只对应一个消费者或者用 routing key 将同一订单的消息路由到同一队列并开启单消费者在消费者内部串行处理。RabbitMQ 本身不保证全局有序但可以保证单队列内有序。堆积了几百万条消息怎么办先查消费者是否在线、是否有异常临时紧急处理可以有几种典型做法扩展消费者并发、加机器修复消费者让消费速度提升紧急情况下甚至可以写脚本直接从队列里把消息转存到数据库再慢慢回放。延迟队列怎么实现TTL 死信交换机具体可参考前面代码。7. 一些个人的实操心得项目用到第三年我开始意识到 RabbitMQ 最大的坑其实不在 RabbitMQ 本身而在消息语义的设计上。如果你的团队还没有统一消息事件的定义格式、日志链路追踪方式消息一多起来就会很痛苦。我从项目里沉淀下来几条个人的心得分享给正在使用或准备使用 RabbitMQ 的朋友第一所有消息必须有统一的事件 ID并且贯穿始终。生产者的 CorrelationData、消费者 Redis 幂等键、日志链路里的 traceId都建议使用同一个 ID。这样一条消息从投递到消费到业务落库全程可以串联查询。第二队列和交换机尽量按照业务模块命名并且把 routing key 的设计当成 API 设计一样认真。比如“预约服务”相关的路由键建议统一格式为 order.created、order.cancelled、order.reminder。从路由键的名称就能看出业务语义避免时间长了之后连写这段代码的人都忘了某个 key 代表什么意思。第三消费者代码里尽量加上注解 RabbitListener 的 concurrency 参数时把这个参数放在常量类里统一管理不要散落到各个方法上。同一套环境不同消费者的并发数应该尽量一致避免某个消费者并发过高把数据库打满。第四存储消息内容和业务数据的数据库尽量分开。不要让消费者直接把队列消费和主要的订单库写操作混在一起否则消费高峰期的数据库负载会影响 Web 请求的稳定性。如果团队资源有限不准备单独建库至少要把消费者处理核心逻辑放到独立的 Service 里和 Controller 层读写保持一定的隔离。写到这里关于 Spring Boot 集成 RabbitMQ 的主要技术点和实战经验基本都覆盖了。最后再分享一个我最近实践的小技巧如果你的项目里同时有多个 RabbitListener 监听多个队列但不同队列的优先级不同可以定制多个 RabbitListenerContainerFactory给不同队列设置不同的 prefetch 和 concurrency 参数。比如订单队列并发设高一些通知队列并发设低一些这样资源分配更合理。不过这属于基线配置之后的调优手段刚刚接触 RabbitMQ 时还是从默认配置跑通再逐步调优比较稳妥。