企业官网建设流程全解析

1. 为什么今天还要认真学 RabbitMQ？不是 Kafka 更火吗？

Spring Boot 实践系列写到第十一章，绕不开消息中间件这个“系统黏合剂”。很多人看到标题第一反应是：RabbitMQ 还值得花时间深挖吗？Kafka 不是早就成了大数据场景的标配？RocketMQ 在金融级事务里也频频露脸——那 RabbitMQ 到底在解决什么别人不好替代的问题？我带团队做过 7 个中大型 Spring Boot 微服务项目，其中 5 个用 RabbitMQ 做核心异步解耦，2 个在初期选了 Kafka 后又切回 RabbitMQ。不是跟风，而是踩过坑之后的真实选择。

核心关键词就三个：可靠投递、复杂路由、轻量嵌入。Kafka 强在高吞吐、日志式持久化、流式处理能力，但它默认不保证单条消息的严格顺序（需分区绑定）、重试机制要自己兜底、消费者位点管理复杂；RocketMQ 对事务消息支持好，但部署依赖 NameServer + Broker 集群，运维成本明显高于单节点 RabbitMQ。而 RabbitMQ 的 AMQP 协议天然支持确认机制（publisher confirm + consumer ack）+ 死信队列（DLX）+ 延迟插件（rabbitmq-delayed-message-exchange）+ 精确到 routing key 的 Topic Exchange 路由——这些能力组合起来，恰恰是订单超时关单、支付结果异步通知、用户行为埋点分发、库存扣减失败补偿等典型业务场景最需要的“确定性”。

更关键的是，Spring Boot 对 RabbitMQ 的封装成熟度极高。@RabbitListener注解一行就能监听队列，RabbitTemplate发送消息连 connection factory 都不用手动 new，自动重连、连接池、JSON 序列化、重试模板（RetryTemplate）全内置。你不需要懂 Erlang，也不用调优 BEAM VM 内存参数，只要理解 Exchange/Queue/Binding 三要素和四种 Exchange 类型的本质区别，就能把 80% 的业务消息流跑稳。我见过太多团队在 Kafka 上卡在“消费者组 rebalance 导致重复消费”或“log compaction 配置错误丢数据”，而 RabbitMQ 在 Spring Boot 场景下，一个spring.rabbitmq.listener.simple.acknowledge-mode=manual加上 try-catch 手动 ack，问题边界非常清晰。

所以这章不讲“RabbitMQ 是什么”，而是聚焦：在 Spring Boot 工程里，怎么让 RabbitMQ 真正扛住生产流量、不出错、可追溯、易扩展。下面所有内容，都来自某电商中台项目真实落地过程——从本地开发调试，到灰度发布压测，再到线上故障排查，每一步都带着血泪教训。

2. 整体设计思路：为什么不用默认配置？为什么必须拆分 Exchange？

2.1 默认配置的三大隐形陷阱

很多新手直接加依赖、配spring.rabbitmq.host就开干，结果上线后出问题才懵。我整理了 Spring Boot Starter 默认配置下最常踩的三个坑：

• 连接池耗尽：CachingConnectionFactory默认cacheMode=CHANNEL，每个线程独占一个 Channel，但 Channel 数量上限默认是 25（channelCacheSize=25）。当并发发送消息超过 25 条/秒，后续请求会阻塞在getConnection()，表现就是接口 RT 突增，CPU 却不高。这不是 RabbitMQ 挂了，是客户端卡死了。

• 消息丢失黑洞：spring.rabbitmq.publisher-confirms=true开启后，若没配RabbitTemplate.setConfirmCallback()，确认失败的消息会静默丢弃，日志里连 warning 都没有。某次支付回调失败，就是因为网络抖动导致 confirm 失败，但代码没监听回调，钱付了却没更新订单状态。

• 死信队列形同虚设：x-dead-letter-exchange参数必须在声明 Queue 时显式设置，且目标 Exchange 必须已存在。Spring Boot 的@Bean Queue queue()方式声明队列时，若没用@Bean Exchange dlxExchange()先声明死信交换机，启动会报错；但若用@RabbitListener自动创建队列，又无法指定 DLX 参数——这就逼你必须手写QueueBean。

所以我们的设计原则很明确：所有关键参数显式声明，拒绝隐式默认；所有异常路径可监控，拒绝静默失败；所有消息生命周期可追溯，拒绝黑盒处理。

2.2 Exchange 拆分策略：按业务域+消息类型二维划分

很多项目只建一个topic.exchange，所有消息往里扔，靠 routing key 区分。这在小项目里没问题，但一旦业务变复杂，问题就来了：

• 运维无法按业务隔离流量，促销活动大促时，商品上架消息洪峰可能挤占支付消息的带宽；
• 某个业务方改了 routing key 格式（比如从order.created.v1改成order.create.v2），其他监听者收不到消息却毫无感知；
• 死信队列无法按业务分类，所有失败消息混在一个 DLQ 里，排查时得 grep 一堆日志。

我们最终采用“业务域前缀 + 消息语义后缀”的 Exchange 命名规范，并为每个核心业务域单独声明 Exchange：

提示：fanout Exchange 不走 routing key 匹配，性能最高，适合“广播类”消息；direct Exchange 要求 routing key 完全匹配，适合“指令类”消息；topic Exchange 支持*和#通配符，适合“事件类”消息。别为了省事全用 topic，通配符越多，RabbitMQ 路由开销越大。

这种拆分带来两个硬性好处：一是运维可通过 RabbitMQ Management UI 单独查看每个 Exchange 的 Publish/Consume Rate，大促时一眼看出哪个业务流量异常；二是某个 Exchange 出问题（如被误删），不影响其他业务，故障域被严格控制。

2.3 消息模型设计：为什么坚持“事件即事实”，拒绝“命令即操作”

这是架构层面的关键取舍。早期我们曾用 RabbitMQ 发送updateInventoryCommand这样的消息，让消费者直接执行 SQL 更新库存。结果出现严重问题：

• 消息重发时，库存被扣减两次；
• 消费者处理慢，大量消息堆积，DB 连接池被打满；
• 无法追溯“谁在什么时候触发了这次扣减”。

后来全部改为“事件即事实”模型：只发送InventoryDeductedEvent，里面包含orderId,skuId,quantity,deductTime,traceId等不可变字段。消费者收到后，先查当前库存是否充足（幂等校验），再执行扣减，最后记录操作日志。这样：

• 重发消息不会导致业务状态错乱（因为每次都是基于当前状态做决策）；
• 扣减逻辑可独立压测，不耦合在消息链路里；
• 所有事件天然带 traceId，ELK 日志一查到底。

Spring Boot 实现上，我们定义统一事件基类：

public abstract class BaseEvent implements Serializable { private static final long serialVersionUID = 1L; private String eventId = UUID.randomUUID().toString(); private String traceId = MDC.get("traceId"); // 从 SLF4J MDC 拿链路 ID private long eventTime = System.currentTimeMillis(); private String sourceService; // 发送方服务名，自动注入 }

所有具体事件继承它，序列化用 Jackson，@JsonInclude(JsonInclude.Include.NON_NULL)避免空字段污染。

3. 核心细节解析：从连接池到死信，每个参数都有它的脾气

3.1 连接工厂：CachingConnectionFactory 的 5 个关键参数

CachingConnectionFactory是 Spring AMQP 的核心，它管理 Connection 和 Channel 的缓存。默认配置在生产环境必然出事，必须显式配置：

spring: rabbitmq: host: ${RABBITMQ_HOST:localhost} port: ${RABBITMQ_PORT:5672} username: ${RABBITMQ_USER:guest} password: ${RABBITMQ_PASS:guest} virtual-host: /prod # 关键：Connection 级别配置 connection-timeout: 30000 # 连接超时 30s，避免 SYN Flood # 关键：Channel 级别配置（这才是性能瓶颈） cache: channel: size: 50 # 每个 Connection 缓存 50 个 Channel，根据并发量调 checkout-timeout: 10000 # 获取 Channel 超时 10s，超时抛异常而非死等 # 关键：Publisher Confirm 配置 publisher-confirms: true publisher-returns: true # 开启 return callback，消息路由失败时触发

重点解释checkout-timeout：当所有缓存 Channel 都被占用，新请求会等待可用 Channel。若设为 -1（无限等待），高并发下线程全卡在这里，Tomcat 线程池耗尽，整个服务不可用。设为 10s，超时后抛AmqpTimeoutException，上层可降级（如记录日志后返回“稍后重试”），保住服务可用性。

注意：channel.size不是越大越好。每个 Channel 占用约 1MB 堆内存，50 个就是 50MB。我们通过压测发现：当 QPS 达到 300 时，channel.size=30就够用；盲目设到 100，内存浪费且 GC 压力增大。

3.2 消息发送：RabbitTemplate 的 3 层防护

RabbitTemplate是发送消息的门面，但直接convertAndSend()风险极大。我们构建了三层防护：

第一层：强制设置 mandatory=true

@Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template = new RabbitTemplate(connectionFactory); template.setMandatory(true); // 关键！路由失败时触发 return callback template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { log.error("Message returned: exchange={}, routingKey={}, replyCode={}, replyText={}", exchange, routingKey, replyCode, replyText); // 这里可发告警、存 DB、推企业微信 }); return template; }

mandatory=true是生死线。它确保：如果消息发到一个不存在的 Exchange，或 routing key 匹配不到任何 Queue，RabbitMQ 不会静默丢弃，而是触发ReturnCallback。我们曾因 Exchange 名字拼错（exchange.order.event写成exchange.order.even），导致 2 小时订单创建事件全丢，就因为没开 mandatory。

第二层：ConfirmCallback 确保投递成功

template.setConfirmCallback((correlationData, ack, cause) -> { if (ack) { log.debug("Message confirmed, id={}", correlationData.getId()); } else { log.error("Message rejected, id={}, cause={}", correlationData.getId(), cause); // 这里必须重发！但注意：重发前要查 DB 确认原消息是否已处理 // 我们用 Redis 记录已发送消息 ID，5 分钟内相同 ID 不重发 } });

correlationData是你传入的唯一标识，必须包含业务 ID（如 orderId）和时间戳，方便对账。ack=false时，cause字段会给出原因，常见如"connection closed"或"channel error"。

第三层：JSON 序列化防坑

默认SimpleMessageConverter只支持String和Serializable，但Serializable有兼容性风险（JDK 版本升级可能反序列化失败）。我们强制用 Jackson：

@Bean public MessageConverter jackson2MessageConverter() { DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); typeMapper.setTypePrecedence(TypePrecedence.TYPE_ID); // 优先用 @class 字段 // 关键：禁用 @class 字段，防止反序列化漏洞 typeMapper.setTrustedPackages("com.example.*"); MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter(); converter.setClassMapper(typeMapper); return converter; }

setTrustedPackages是安全红线。不设它，攻击者可构造恶意 JSON，反序列化任意类执行命令。只信任自己包下的类，彻底堵死漏洞。

3.3 消息接收：@RabbitListener 的 4 个致命细节

@RabbitListener看似简单，但 90% 的线上问题出在这里：

细节 1：并发消费者数不是越多越好

@RabbitListener( queues = "queue.order.created", concurrency = "3-10", // 最小 3，最大 10 个消费者线程 containerFactory = "simpleRabbitListenerContainerFactory" ) public void onOrderCreated(OrderCreatedEvent event) { // 处理逻辑 }

concurrency="3-10"表示 Spring 会动态调整消费者数量。但注意：每个消费者线程独占一个 Channel。如果channel.size=30，最多只能起 30 个消费者；若设concurrency="1-50"，实际最多只有 30 个生效，剩下 20 个永远起不来。我们压测发现：当数据库连接池大小为 20 时，消费者并发数设到 25 就开始报HikariPool-1 - Connection is not available，因为每个消费者都在抢 DB 连接。最终定为concurrency="1-8"，配合 DB 连接池maximum-pool-size=16，资源利用率最稳。

细节 2：手动 ACK 必须包裹完整业务逻辑

@RabbitListener(queues = "queue.payment.result") public void onPaymentResult(Message message, Channel channel, @Payload PaymentResultEvent event) throws IOException { try { // 1. 处理业务逻辑（更新订单、发通知等） orderService.handlePaymentResult(event); // 2. 业务成功，才手动 ACK channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (Exception e) { log.error("Failed to process payment result", e); // 3. 业务失败，拒绝消息并进入死信队列 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); } }

关键点：basicAck()必须在业务逻辑完全成功后调用。如果orderService.handlePaymentResult()里抛了异常，但basicAck()已执行，消息就丢了。我们曾因此漏处理 3 笔大额支付，根源就是 ACK 放错了位置。

细节 3：死信队列（DLQ）必须显式声明

@Bean public Queue orderCreatedQueue() { Map<String, Object> args = new HashMap<>(); args.put("x-dead-letter-exchange", "exchange.dlx"); // 死信交换机 args.put("x-dead-letter-routing-key", "dlq.order.created"); // 死信路由键 args.put("x-message-ttl", 300000); // 消息存活 5 分钟，超时进 DLQ return QueueBuilder.durable("queue.order.created") .withArguments(args) .build(); } @Bean public Exchange dlxExchange() { return ExchangeBuilder.topicExchange("exchange.dlx").durable(true).build(); } @Bean public Queue dlqOrderCreatedQueue() { return QueueBuilder.durable("queue.dlq.order.created").build(); } @Bean public Binding dlqBinding() { return BindingBuilder.bind(dlqOrderCreatedQueue()) .to(dlxExchange()) .with("dlq.order.created") .noargs(); }

这里有两个易错点：

• x-dead-letter-exchange必须是已存在的 Exchange，所以dlxExchange()Bean 必须在orderCreatedQueue()之前初始化（Spring Boot 默认按字母序，dlxExchange字母比orderCreatedQueue靠前，所以 OK）；
• x-message-ttl是毫秒单位，别写成300000ms，RabbitMQ 会忽略。

细节 4：消息重试要配退避策略，不能指数爆炸

Spring Boot 默认重试是立即重试 3 次，这对网络抖动有效，但对 DB 连接失败无效——重试 3 次还是连不上。我们用RetryTemplate配置退避：

@Bean public RetryTemplate retryTemplate() { RetryTemplate template = new RetryTemplate(); // 重试策略：最大重试 3 次（含首次） SimpleRetryPolicy policy = new SimpleRetryPolicy(); policy.setMaxAttempts(3); template.setRetryPolicy(policy); // 退避策略：第一次延时 1s，第二次 3s，第三次 9s（3 的幂次） ExponentialBackOffPolicy backOff = new ExponentialBackOffPolicy(); backOff.setInitialInterval(1000); backOff.setMultiplier(3.0); backOff.setMaxInterval(9000); template.setBackOffPolicy(backOff); return template; }

ExponentialBackOffPolicy是精髓。线性退避（每次+1s）在故障恢复时太慢；固定退避（每次都等 5s）又太激进。指数退避在“快速失败”和“耐心等待”间取得平衡。

4. 实操过程：从本地调试到线上灰度，每一步都踩过坑

4.1 本地开发：用 Docker Compose 搭建最小可用环境

本地不能连测试环境 RabbitMQ，必须自建。但docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:3.11-management有个大坑：默认虚拟主机是/，而 Spring Boot 默认连/，但生产环境我们用/prod。如果本地不统一，代码里写死virtual-host=/prod，本地就连不上。

解决方案：Docker Compose 统一配置：

# docker-compose.yml version: '3.8' services: rabbitmq: image: rabbitmq:3.11-management ports: - "5672:5672" - "15672:15672" environment: RABBITMQ_DEFAULT_VHOST: /dev RABBITMQ_DEFAULT_USER: devuser RABBITMQ_DEFAULT_PASS: devpass volumes: - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf

rabbitmq.conf文件内容：

# 启用延迟消息插件（必须！） plugins = rabbitmq_delayed_message_exchange # 设置默认虚拟主机 default_vhost = /dev # 开启 MQTT、STOMP 等协议（可选） # listeners.tcp.default = 5672

然后 Spring Boot 配置：

spring: profiles: dev rabbitmq: virtual-host: /dev username: devuser password: devpass

这样本地和测试环境配置完全一致，切换 profile 就行，杜绝“本地 OK，测试环境挂”的低级错误。

4.2 消息追踪：如何在海量日志中定位一条消息？

大促时每秒几万条消息，出问题怎么查？我们建立三级追踪体系：

第一级：消息 ID 全链路透传

发送端：

CorrelationData correlationData = new CorrelationData( String.format("order-%s-%d", event.getOrderId(), System.currentTimeMillis()) ); rabbitTemplate.convertAndSend( "exchange.order.event", "order.created", event, correlationData );

消费端：

@RabbitListener(queues = "queue.order.created") public void onOrderCreated(@Payload OrderCreatedEvent event, Message message, Channel channel) { // 从消息属性里提取原始消息 ID String originalId = message.getMessageProperties() .getHeaders() .getOrDefault("__TypeId__", "") .toString(); // Spring AMQP 自动加的类型头 MDC.put("msgId", originalId); try { orderService.createOrder(event); channel.basicAck(...); } finally { MDC.clear(); // 清理 MDC，避免线程复用污染 } }

第二级：RabbitMQ Management UI 实时监控

打开http://localhost:15672，进Exchanges页面，点exchange.order.event，看Publish rate和Publish details。如果Publish rate突降为 0，说明上游服务挂了；如果Publish details里Unroutable messages数字上涨，说明 routing key 配错了。

第三级：ELK 日志关联查询

所有服务日志都打traceId和msgId。在 Kibana 里搜msgId: "order-123456-1712345678900"，就能看到这条消息从发送、路由、消费、ACK 的全过程日志。我们甚至写了脚本，输入订单号，自动查出对应的所有消息轨迹。

4.3 灰度发布：如何验证新版本消费者不丢消息？

上线新版本消费者（比如修复了库存超卖 bug），最怕旧消息被新代码错误处理。我们采用“双写 + 对账”灰度法：

1. 新版本消费者监听queue.order.created.v2（新队列），旧版本监听queue.order.created（老队列）；
2. 发送端同时发两条消息：一条到老 Exchange（exchange.order.event），一条到新 Exchange（exchange.order.event.v2），内容完全一样；
3. 用 Flink 实时计算两个队列的消费成功率，偏差超过 0.1% 就告警；
4. 稳定运行 24 小时后，切流：把老 Exchange 的 Binding 全删掉，新 Exchange 的 Binding 全加上。

实操心得：切流前必须停掉所有生产者！否则新旧消息混发，对账就乱了。我们用 Apollo 配置中心控制开关，切流时一键关闭order-service的消息发送功能，等队列清空后再开启。

4.4 生产压测：用 Gatling 模拟真实流量

不能只压接口，要压消息链路。我们用 Gatling 写脚本，模拟用户下单：

class RabbitMQSimulation extends Simulation { val rabbitmqConf = rabbitmq("amqp://devuser:devpass@localhost:5672/dev") .connectionFactory(new CachingConnectionFactory()) val scn = scenario("Order Creation") .exec(rabbitmq("Send Order Event") .publish("exchange.order.event", "order.created") .message(ElFileBody("src/test/resources/order-event.json")) .headers(Map("content-type" -> "application/json")) ) setUp(scn.inject(atOnceUsers(1000))).protocols(rabbitmqConf) }

压测发现两个关键指标：

• 当channel.size=20时，1000 并发下publish rate稳定在 800 msg/s，但unacknowledged消息数持续上涨，说明消费者处理不过来；
• 把消费者concurrency从1-5调到1-12，unacknowledged归零，publish rate提升到 1200 msg/s。

这验证了我们前面说的：Channel 数量和消费者并发数必须协同调优，不能只看单点。

5. 常见问题与排查技巧实录：那些凌晨三点的电话

5.1 消息堆积：不是 RabbitMQ 慢，是消费者卡住了

现象：Management UI 显示queue.order.created的Ready消息数从 0 暴涨到 50 万，Unacknowledged为 0。

排查步骤：

1. curl -u devuser:devpass http://localhost:15672/api/queues/%2Fdev/queue.order.created查队列详情，重点关注messages_ready和consumers字段；
2. 如果consumers=0，说明消费者进程挂了或没连上——检查消费者服务日志，看是否有Connection refused；
3. 如果consumers=1但messages_ready持续涨，说明消费者处理慢——用jstack <pid>看线程栈，大概率卡在 DB 查询或远程 HTTP 调用；
4. 临时方案：重启消费者服务；长期方案：给 DB 查询加索引，HTTP 调用加熔断。

独家技巧：在消费者里加心跳上报。每 30 秒往 Redis 写一次consumer:order-created:last-active时间戳，Prometheus 定期拉取，超 60 秒没更新就告警。比单纯看进程存活更精准。

5.2 消息重复：不是 RabbitMQ 问题，是 ACK 时机错了

现象：同一笔订单，库存被扣减了两次，日志里看到两条order-123456的处理记录。

根因分析：

• 消费者处理业务花了 15 秒（超了 RabbitMQ 的heartbeat=60s），RabbitMQ 认为 Channel 断连，把消息重新入队；
• 消费者线程其实还在跑，最后basicAck()成功，但消息已二次入队。

解决方案：

1. 业务层幂等：订单号 + 操作类型（DEDUCT_STOCK）作为唯一键，插入前INSERT IGNORE；
2. 基础设施层优化：调大spring.rabbitmq.listener.simple.heartbeat=120，并确保消费者处理逻辑在 60 秒内完成；
3. 终极方案：用@RabbitListener的acknowledge-mode=MANUAL+default-requeue-rejected=false，失败消息不重入队，直接进 DLQ。

5.3 连接频繁断开：SSL 配置引发的血案

现象：日志里大量Connection reset by peer，spring.rabbitmq.connection-timeout设为 30s 也没用。

真相：测试环境 RabbitMQ 开了 SSL，但 Spring Boot 配置里没开spring.rabbitmq.ssl.enabled=true，客户端用明文连，服务端直接 reset。

修复方法：

• 查 RabbitMQ 日志：/var/log/rabbitmq/rabbit@xxx.log，搜索SSL关键字；
• 若看到SSL handshake failed，立刻加配置：

spring: rabbitmq: ssl: enabled: true algorithm: TLSv1.2 key-store: classpath:rabbitmq-keystore.jks key-store-password: changeit trust-store: classpath:rabbitmq-truststore.jks

• 生成 keystore 用keytool -import -alias rabbitmq -file rabbitmq.crt -keystore rabbitmq-keystore.jks。

5.4 死信队列不工作：90% 是 Exchange 不存在

现象：消息 TTL 到期，但没进 DLQ，Management UI 里queue.dlq.order.created的Ready一直是 0。

检查清单：

• ✅x-dead-letter-exchange值是否和@Bean Exchange dlxExchange()名字完全一致（大小写敏感）；
• ✅dlxExchange()Bean 是否被 Spring 扫描到（加@Configuration）；
• ✅x-dead-letter-routing-key是否和Binding的with()参数一致；
• ❌ 最容易漏：dlxExchange()的durable(true)是否设置？非持久化 Exchange 在 RabbitMQ 重启后消失，DLQ 就失效。

常见问题速查表：

5.5 性能瓶颈定位：用 rabbitmqctl 看清真相

当 Management UI 图表不够用，直接上命令行：

• 查所有连接：rabbitmqctl list_connections，看state是否大量running（正常）或closing（异常）；
• 查 Channel 状态：rabbitmqctl list_channels，关注user,connection,state，state=closing说明 Channel 泄漏；
• 查队列内存占用：rabbitmqctl list_queues name memory messages_ready，memory超 100MB 就危险，可能 OOM；
• 强制清理队列：rabbitmqctl purge_queue queue.order.created（慎用！只在测试环境）。

我们曾用list_channels发现一个 Bug：消费者处理异常时没关 Channel，导致 200+ Channel 卡在closing状态，占满连接池。修复后，channel.size从 50 降到 20，性能反而提升。

6. 最后分享一个小技巧：用 Spring Boot Actuator 暴露 RabbitMQ 健康指标

Spring Boot Actuator 默认只暴露rabbit健康端点（/actuator/health/rabbit），但它是二值的（UP/DOWN），不够细。我们扩展了自定义端点，暴露关键指标：

@Component @Endpoint(id = "rabbitmq-stats") public class RabbitMQStatsEndpoint { @Autowired private RabbitAdmin rabbitAdmin; @ReadOperation public Map<String, Object> getStats() { Map<String, Object> stats = new HashMap<>(); try { // 查队列消息数 stats.put("queue_order_created_ready", rabbitAdmin.getQueueProperties("queue.order.created") .get("QUEUE_MESSAGES_READY")); // 查连接数 stats.put("connections_count", rabbitAdmin.getRabbitTemplate().getConnectionFactory() .createConnection().getChannelCount()); } catch (Exception e) { stats.put("error", e.getMessage()); } return stats; } }

配置application.yml：

management: endpoints: web: exposure: include: health,metrics,rabbitmq-stats

访问http://localhost:8080/actuator/rabbitmq-stats，就能看到实时队列深度。我们把它接入 Grafana，画成折线图，阈值告警——这才是真正的可观测性。

我在实际使用中发现，所有“高级功能”都建立在基础配置正确之上。与其研究 RabbitMQ 的镜像队列、联邦插件，不如先把channel.size、mandatory、DLX这三件事做扎实。消息中间件不是炫技的舞台，而是业务稳定的基石。某次大促零故障，不是因为我们用了多牛的技术，而是因为凌晨三点排查时，能快速定位到是concurrency配少了——这种确定性，才是工程师最该追求的东西。