Knowledge4J

切换主题

08 - SpringBoot 整合 RabbitMQ ​

面试高频度：⭐⭐⭐⭐⭐ 考查方式：实操题、代码设计题、踩坑经验题

一、Spring AMQP 是什么？ ​

1.1 原生 AMQP API 的问题 ​

如果直接使用 RabbitMQ 的 Java 客户端（com.rabbitmq:amqp-client），你需要手动管理 Connection、Channel、声明 Exchange/Queue/Binding、处理 ACK、处理异常……代码量大且容易出错。

Spring AMQP 是 Spring 生态对 AMQP 协议的封装，核心价值：

  • 自动管理连接和 Channel（连接池、Channel 缓存）
  • 声明式配置（注解驱动，不需要手写 channel.queueDeclare()）
  • 自动序列化/反序列化（JSON 转换器，消息体自动转对象）
  • 内置重试机制（配置化重试策略，不需要手写重试逻辑）
  • 与 Spring Boot 深度整合（application.yml 配置，开箱即用）

1.2 核心组件 ​

组件作用对应原生 API
RabbitTemplate发消息的工具类Channel.basicPublish()
@RabbitListener消费消息的注解Channel.basicConsume() + DeliverCallback
Queue、Exchange、Binding声明式 BeanChannel.queueDeclare() / exchangeDeclare() / queueBind()
MessageConverter消息序列化/反序列化手动 getBytes() / new String()

二、配置文件——每个配置项的含义 ​

yaml
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

连接池配置：

yaml
    cache:
      channel:
        size: 50            # 缓存的 Channel 数量
      connection:
        mode: channel       # channel 模式（共享连接）或 connection 模式（每个线程独立连接）

为什么要缓存 Channel？因为每次发送/消费消息都需要 Channel，如果每次都创建新 Channel，开销很大。缓存一批 Channel 复用，减少创建销毁的频率。size=50 表示最多缓存 50 个 Channel——如果并发超过 50，超出的线程需要等待有 Channel 空闲。

生产端配置：

yaml
    publisher-confirm-type: correlated   # Confirm 模式
    publisher-returns: true              # Return 回调
配置项值含义
publisher-confirm-typenone不开启 Confirm
correlated异步 Confirm，通过 correlationId 关联消息
simple同步 Confirm（性能差，不推荐）
publisher-returnstrue开启 Return 回调，路由失败时通知 Producer

消费端配置：

yaml
    listener:
      simple:
        acknowledge-mode: manual         # 手动 ACK
        prefetch: 10                     # 预取数量
        concurrency: 5                   # 最小消费者线程数
        max-concurrency: 20              # 最大消费者线程数
配置项含义为什么这么设
acknowledge-mode: manual手动 ACK生产必须——自动 ACK 会丢消息
prefetch: 10每个线程最多持 10 条未 ACK 消息太大内存压力大，太小吞吐低
concurrency: 5启动时创建 5 个消费者线程保证基础消费能力
max-concurrency: 20最多扩展到 20 个线程根据消息积压情况动态扩缩

三、声明 Exchange / Queue / Binding ​

3.1 配置类方式（推荐） ​

为什么推荐配置类而不是注解声明？ 配置类方式把所有基础设施（Exchange、Queue、Binding、死信配置）集中在一个地方，代码清晰、易维护、易修改。注解声明虽然简洁，但配置分散在消费类中，不易管理。

声明顺序很重要：先声明死信 Exchange 和 Queue，再声明正常 Queue（因为正常 Queue 要绑定到死信 Exchange）。Spring AMQP 会自动处理 Bean 的依赖关系，但逻辑上要清楚这个依赖链。

关键配置项说明：

配置含义不设的后果
.durable(true)Queue/Exchange 持久化Broker 重启后消失
.deadLetterExchange()绑定死信交换机消费失败的消息无处可去
.deadLetterRoutingKey()死信路由键死信消息可能路由到错误的队列
.maxLength()Queue 最大消息数无法限制队列长度，可能 OOM
.ttl()消息级 TTL消息永不过期（如果业务需要过期的话）

3.2 RabbitTemplate 配置——发送端的可靠性 ​

RabbitTemplate 是 Spring AMQP 的发送核心。配置 Confirm 和 Return 回调是保证发送可靠性的关键：

配置作用对应的原生机制
setConfirmCallbackBroker 确认消息到达 Exchange 时回调channel.addConfirmListener()
setReturnsCallback消息从 Exchange 路由到 Queue 失败时回调channel.addReturnListener()
setMandatory(true)配合 Return 使用，路由失败时返回消息basicPublish(exchange, key, true, ...)

注意：setMandatory(true) 必须设置，否则路由失败时消息会被静默丢弃，Return 回调不会触发。

3.3 MessageConverter——消息序列化 ​

默认情况下 Spring AMQP 使用 Java 序列化（SimpleMessageConverter），生成的消息体是二进制 Java 对象——跨语言不友好、体积大、有安全风险。

生产推荐：Jackson2JsonMessageConverter，消息体为 JSON 格式——体积小、可读性好、跨语言兼容。

四、消费者——手动 ACK + 异常处理 ​

4.1 为什么必须手动 ACK？ ​

Spring AMQP 的 @RabbitListener 在 acknowledge-mode: auto（默认）时，消费方法执行完就自动 ACK，执行异常就自动 NACK+requeue。这看似方便，但有两个问题：

  • 方法执行成功不代表业务成功——方法可能执行了一半抛异常，但 Spring 已经 ACK 了
  • 自动 NACK+requeue 会死循环——如果失败原因是代码 Bug，消息会反复入队、反复失败

4.2 手动 ACK 的正确模式 ​

正确的消费流程：

1. 解析消息
2. 幂等校验（Redis / 唯一约束）
   → 重复消息 → basicAck → return
3. 处理业务逻辑
   → 成功 → basicAck
   → 失败 → 检查重试次数
     → 未超限 → basicNack(requeue=true)
     → 已超限 → basicNack(requeue=false) → 进入 DLX

关键细节：

  • Channel 对象由 Spring AMQP 自动注入到 @RabbitListener 方法参数中，不需要手动创建
  • deliveryTag 通过 message.getMessageProperties().getDeliveryTag() 获取
  • basicAck(tag, false) 中的 false 表示只确认当前这一条（true 表示确认该 tag 及之前的所有消息）

4.3 批量消费 ​

当消息量很大时，逐条消费的开销（每条消息一次网络往返 + 一次 ACK）太高。批量消费可以显著提升吞吐量：

批量消费的原理：Spring AMQP 一次从 Queue 取多条消息，打包成 List<Message> 传给消费方法。处理完后用最后一条消息的 deliveryTag + multiple=true 做批量 ACK。

批量消费的前提：需要在 SimpleRabbitListenerContainerFactory 中配置 batchSize 和 consumerBatchEnabled。

批量消费的权衡：

  • 优势：减少网络往返次数、减少 ACK 次数、可以批量写数据库
  • 劣势：一条消息失败会导致整批重试、消费延迟可能增大（要等攒够一批）

五、延迟队列配置 ​

Spring AMQP 对延迟队列插件的支持通过 CustomExchange 实现：

为什么是 CustomExchange？ 因为 x-delayed-message 不是 AMQP 协议内置的 Exchange 类型，是插件自定义的。Spring AMQP 提供了 CustomExchange 来声明任意类型的 Exchange。

六、常见踩坑点 ​

踩坑根因解决方案
消息发送成功但队列为空RoutingKey 写错或未绑定开启 Return 回调，监听路由失败
消费者收不到消息prefetch 太小导致消息被其他消费者"抢走"；或 Channel 未正确创建检查 prefetch 和 concurrency 配置
消费者线程阻塞消费方法中同步调用下游服务（DB/API），下游超时导致线程被占用设置超时 + 异步化（消费线程只提交任务，不做业务）
消息丢失未配持久化 / 使用自动 ACK三重持久化 + 手动 ACK（见 **消息可靠性保障**）
重复消费未实现幂等Redis 去重 + 状态机（见 **消息幂等性与重复消费**）
连接泄漏Channel 异常后未正确关闭使用 Spring AMQP 的连接池管理，不要手动创建 Connection/Channel
内存溢出prefetch 太大（大量消息堆积在消费者内存）或消息体过大prefetch=10-50，消息体 < 1MB
@RabbitListener 方法抛异常后消息一直重试默认行为是 NACK+requeue，消息回到队列又被消费，无限循环手动 ACK + 重试计数 + 超限发 DLX

七、最佳实践清单 ​

生产端：

  • ✅ Confirm 模式（correlated 异步）
  • ✅ Mandatory + Return 回调
  • ✅ 消息持久化（deliveryMode=PERSISTENT）
  • ✅ 消息 ID（业务唯一键，用于幂等去重）
  • ✅ 消息体 < 1MB

消费端：

  • ✅ 手动 ACK
  • ✅ prefetch 10-50
  • ✅ 幂等校验（Redis + 状态机 + DB 唯一约束）
  • ✅ 异常处理（有限次重试 + 死信队列兜底）
  • ✅ 消费超时控制

运维端：

  • ✅ Exchange + Queue 持久化
  • ✅ 仲裁队列（Quorum Queue）
  • ✅ 内存/磁盘告警
  • ✅ 消息积压告警（messages_ready > 1万 → P2）
  • ✅ 连接/Channel 数监控

八、关联笔记 ​

  • **交换机类型与路由策略** - Exchange 声明细节
  • **消息可靠性保障** - Confirm/Return/ACK 的底层原理
  • **死信队列与延迟队列** - 延迟队列的原理和方案对比
  • **RabbitMQ面试高频题汇总** - Spring 相关面试题

Knowledge4J — Java 知识库

MIT License