Spring + 消息队列实战：从异步解耦到可靠投递

关键词：Spring Boot / RabbitMQ / Kafka / 消息可靠性 / 幂等消费 / 死信队列

本文以 Spring Boot 3.x + RabbitMQ 为主线，讲清楚消息队列的核心场景、完整代码实现，以及生产环境绕不开的三个问题：消息不丢、消息不重、消息有序。文末附 Spring Kafka 的对应写法与最佳实践清单。

一、为什么需要消息队列？

先看一个所有业务系统都会遇到的场景：用户下单。

同步调用：
下单接口 ──> 扣库存(200ms) ──> 加积分(150ms) ──> 发短信(300ms) ──> 返回用户
                                                        总耗时 650ms+

同步链路有几个致命问题：

• 响应慢：接口耗时是所有下游调用的总和，任一环节变慢，整个接口都变慢；
• 强耦合：积分服务挂了，下单也跟着失败——但"加积分"真的是下单的必要条件吗？
• 扛不住峰值：大促时下单 QPS 暴涨 10 倍，短信服务第一个被打挂，然后雪崩。

引入消息队列后，架构变成这样：

异步化后：
下单接口 ──> 扣库存 ──> 返回用户（200ms）
                │
                └──> MQ ──> 积分服务（异步消费）
                        └──> 短信服务（异步消费）

这就是消息队列的三大经典作用：

但天下没有免费的午餐，引入 MQ 的代价也要心里有数：

• 系统复杂度上升：多了一个需要部署、监控、运维的中间件；
• 一致性变难：主流程成功、消息消费失败怎么办？（后文详述）
• 可用性依赖：MQ 集群本身挂了，所有异步链路都会中断。

一句话结论：非核心链路、允许短暂延迟的业务，才适合消息队列；强一致的核心链路老老实实同步调用。

二、技术选型：RabbitMQ、Kafka、RocketMQ 怎么选

选型没有银弹：

• 中小规模业务系统（订单、通知、积分）：RabbitMQ 上手最快、功能够全，Spring AMQP 集成成熟，是大多数团队的首选；
• 日志、埋点、大数据管道：Kafka 几乎是唯一答案，吞吐无敌；
• 交易类核心链路，需要事务消息和延迟消息：优先看 RocketMQ。

本文以 RabbitMQ + Spring AMQP 为主线讲解，第八节给出 Spring Kafka 的对应写法。

三、快速上手：Spring Boot 集成 RabbitMQ

3.1 环境与依赖

# 启动带管理界面的 RabbitMQ（管理台：http://localhost:15672，guest/guest）
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.13-management

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /
    username: guest
    password: guest
    # ---- 生产者确认（第四节详解）----
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
    # ---- 消费者手动 ACK（第四节详解）----
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 5
        concurrency: 2

3.2 声明交换机、队列与绑定

以"订单创建"为例，我们声明一套业务拓扑，并顺手把死信拓扑也建好（后面会用到）：

@Configuration
public class RabbitConfig {

    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_QUEUE    = "order.queue";
    public static final String ORDER_KEY      = "order.created";

    public static final String DLX_EXCHANGE = "dlx.order.exchange";
    public static final String DLX_QUEUE    = "dlx.order.queue";
    public static final String DLX_KEY      = "order.dead";

    @Bean
    public DirectExchange orderExchange() {
        return ExchangeBuilder.directExchange(ORDER_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
                .deadLetterExchange(DLX_EXCHANGE)   // 消费失败/过期后路由到死信
                .deadLetterRoutingKey(DLX_KEY)
                .build();
    }

    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ORDER_KEY);
    }

    @Bean
    public DirectExchange dlxExchange() {
        return ExchangeBuilder.directExchange(DLX_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue dlxQueue() {
        return QueueBuilder.durable(DLX_QUEUE).build();
    }

    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(DLX_KEY);
    }

    /** 使用 JSON 序列化，替代默认的 JDK 序列化；注入 Spring 的 ObjectMapper 以支持 LocalDateTime 等类型 */
    @Bean
    public MessageConverter jacksonMessageConverter(ObjectMapper objectMapper) {
        return new Jackson2JsonMessageConverter(objectMapper);
    }
}

两个细节值得注意：

1. 默认的 JDK 序列化要求消息类实现 Serializable，且生产/消费两端类路径一致，跨服务传对象非常脆弱，生产环境一律换成 JSON；
2. Jackson2JsonMessageConverter 如果自己 new ObjectMapper()，是不带 JavaTimeModule 的，LocalDateTime 字段会直接报错——所以这里注入 Spring Boot 自动配置好的 ObjectMapper。

3.3 消息体与生产者

public record OrderMessage(
        String messageId,     // 全局唯一 ID，用于消费端幂等去重
        String orderId,
        BigDecimal amount,
        LocalDateTime createdAt
) {}

@Component
@Slf4j
public class OrderProducer {

    private final RabbitTemplate rabbitTemplate;

    public OrderProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        initCallbacks();
    }

    private void initCallbacks() {
        // ConfirmCallback：消息是否成功到达 Broker
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            String msgId = correlationData != null ? correlationData.getId() : "unknown";
            if (ack) {
                log.debug("消息已到达 Broker, id={}", msgId);
            } else {
                log.error("消息未到达 Broker, id={}, cause={}", msgId, cause);
                // TODO: 记录失败消息 -> 落库/告警/补偿重发
            }
        });

        // ReturnsCallback：消息到达了 Broker，但没有路由到任何队列（如 routingKey 写错）
        rabbitTemplate.setReturnsCallback(returned -> log.error(
                "消息未路由到队列: exchange={}, routingKey={}, replyText={}",
                returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()));
    }

    public void sendOrderCreated(OrderMessage message) {
        CorrelationData correlationData = new CorrelationData(message.messageId());
        rabbitTemplate.convertAndSend(
                RabbitConfig.ORDER_EXCHANGE,
                RabbitConfig.ORDER_KEY,
                message,
                correlationData);
    }
}

到这里，一个"能跑"的收发链路就完成了。但如果只做到这一步就上生产，消息迟早会丢。

四、生产环境第一课：消息不能丢

消息从生产到消费，要跨三个环节，每个环节都可能丢：

生产者 ──①──> Broker（MQ 服务器） ──②──> 队列存储 ──③──> 消费者

对应三道防线：生产者确认、Broker 持久化、消费者手动 ACK。

4.1 防线一：生产者确认（Publisher Confirm）

convertAndSend 只是"把消息丢给了网络"，并不能保证 Broker 收到。开启 Confirm 机制后，Broker 会异步回执：

• ConfirmCallback：消息是否成功到达 Broker；
• ReturnsCallback：消息到达了 Broker，但没有路由到任何队列（典型原因：routingKey 配错、队列被删）。

这正是 3.1 yaml 中这三行的作用：

publisher-confirm-type: correlated   # 以关联方式异步回执，代替旧的 publisher-confirms: true
publisher-returns: true
template:
  mandatory: true                    # 无法路由时不静默丢弃，而是触发 ReturnsCallback

两个必须知道的坑：

1. 回调是异步的，发送线程发起 convertAndSend 后立即返回，结果由回调线程送达。所以"发送成功"只能事后知道，不能用来控制主流程；
2. 对可靠性要求极高的场景（如支付结果通知），标准做法是本地消息表：
   • 业务数据和"待发送消息"在同一个本地事务里落库；
   • 事务提交后再真正发送 MQ，发送成功则把消息标记为已完成；
   • 后台定时任务扫描超时未完成的消息，补偿重发。

这样即使"业务成功但发 MQ 失败"，消息也不会凭空消失。

4.2 防线二：Broker 持久化

Broker 重启，内存中的消息会全部丢失。所以交换机、队列、消息三者都要持久化：

• 声明时 durable(true)（QueueBuilder.durable() / ExchangeBuilder...durable(true)，本文示例已全部持久化）；
• 使用 JSON 序列化后，Spring AMQP 默认以 PERSISTENT 模式投递，无需额外设置。

单机持久化只保证"重启不丢"，机器磁盘坏了照样丢。对可靠性有要求的集群，使用 Quorum 队列（RabbitMQ 3.8+，基于 Raft 的多副本队列）代替经典镜像队列。

4.3 防线三：消费者手动 ACK

Spring AMQP 默认自动 ACK：方法一被调用就确认，哪怕下一行业务代码就抛异常——消息没了，业务却没做。生产环境必须改为手动 ACK：

listener:
  simple:
    acknowledge-mode: manual
    prefetch: 5        # 每个消费者最多持有 5 条未确认消息

@Component
@Slf4j
@RequiredArgsConstructor
public class OrderConsumer {

    private final StringRedisTemplate redisTemplate;
    private final OrderService orderService;

    @RabbitListener(queues = RabbitConfig.ORDER_QUEUE)
    public void onOrderCreated(OrderMessage message,
                               Channel channel,
                               @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        log.info("收到订单消息: {}", message);

        // 1. 幂等去重（第五节详解）
        Boolean first = redisTemplate.opsForValue()
                .setIfAbsent("mq:dedup:" + message.messageId(), "1", Duration.ofHours(24));
        if (!Boolean.TRUE.equals(first)) {
            log.warn("重复消息，直接确认丢弃: {}", message.messageId());
            channel.basicAck(deliveryTag, false);
            return;
        }

        try {
            // 2. 业务处理
            orderService.handleOrderCreated(message);
            // 3. 业务成功，才确认消费
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("消费失败", e);
            // 4. 删除去重标记，保证重试时消息还能被处理
            redisTemplate.delete("mq:dedup:" + message.messageId());
            // requeue=false：不重回原队列，转入死信队列
            channel.basicNack(deliveryTag, false, false);
        }
    }
}

三个 API 的语义：

4.4 死信队列：给失败消息一个去处

一条消息变成"死信"有三种情况：

1. 消费端 nack/reject 且 requeue=false；
2. 消息在队列中 TTL 过期；
3. 队列达到最大长度，消息被挤出。

3.2 中我们已经通过 deadLetterExchange 给业务队列绑定了死信拓扑。失败的消息会进入 dlx.order.queue，由专门的消费者接手：

@RabbitListener(queues = RabbitConfig.DLX_QUEUE)
public void onDeadLetter(OrderMessage message, Channel channel,
                         @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    log.error("死信消息: {}", message);
    deadLetterService.saveForCompensation(message);   // 落库 + 告警，人工或定时任务补偿
    channel.basicAck(tag, false);
}

为什么失败要转死信而不是无限 requeue？ 因为一条"毒消息"（永远处理失败）如果无限重回队列，会把整个消费线程堵死。正确姿势是：有限重试 → 进死信 → 落库 → 告警 → 人工/定时补偿。

至此三道防线闭环：

生产者                Broker                      消费者
Confirm/Return  ──>  持久化 + Quorum 队列  ──>  手动 ACK
本地消息表兜底                              失败转死信 + 补偿

五、重复消息与消费幂等

5.1 重复是常态，不是意外

很多文章把"消息不重复"当成可靠性指标之一，这是误导。在"至少一次（At Least Once）"的投递语义下，重复不可避免：

• 生产者发送后 Confirm 回执在网络中超时，生产者重发 → Broker 里两条；
• 消费者处理完成，但 ACK 在网络中丢失，Broker 认为未消费 → 重新投递。

追求"不重复"是徒劳的，工程上的正确目标是：允许重复投递，但业务处理必须幂等——同一条消息处理一次和处理一百次，结果一致。

5.2 三种幂等方案

5.3 实现要点

第四节 OrderConsumer 里的写法就是"Redis SETNX + 手动 ACK"的组合，有三个容易踩坑的细节：

1. 先抢占去重标记，业务失败时要删掉标记，否则第一次消费失败后，重试的消息会被去重逻辑误杀；
2. 先业务后 ACK：如果先 ACK 后业务，业务失败时消息已经确认，无法重试；
3. Redis 去重与业务处理不是原子操作，最严格的场景（如资金类）要用 DB 唯一索引兜底，Redis 只作为前置挡板。

六、顺序消费

6.1 为什么会乱序

队列本身是先进先出的，乱序来自两个地方：

• 生产端：同一订单的"创建 → 支付 → 发货"三条消息，可能被发到了不同队列/不同分区，消费顺序失控；
• 消费端：队列有序，但多个消费者并发消费，谁先处理完说不准。

6.2 RabbitMQ 的方案

核心思路：同一业务 ID 的消息进同一个队列，且该队列只有一个消费者线程。

listener:
  simple:
    prefetch: 1      # 一次只取一条，处理完再取，严格有序
    concurrency: 1   # 该队列单线程消费

发消息时用 orderId 作为 routingKey（或用 Topic 交换机按 orderId 哈希路由），保证同一订单的消息落在同一个队列。

代价是牺牲并发：单队列单线程吞吐有限。实践中的折中是按 orderId 分片到 N 个队列，队列之间并行、队列之内串行——和 Kafka 的分区思想完全一致。

6.3 Kafka 天然的优势

Kafka 只保证分区内有序，所以只要让同一业务 ID 的消息进同一个分区即可：

// 第二个参数是 key：相同 orderId 的消息哈希到同一分区，天然有序
kafkaTemplate.send("order-events", message.orderId(), message);

消费端分区由 Kafka 消费者组分配，每个分区同一时刻只被组内一个线程消费，无需额外处理。

七、延迟消息

经典场景：订单 30 分钟未支付自动取消。轮询订单表既浪费又延迟，用延迟消息优雅得多。

方案一：TTL + 死信队列（能跑，但有坑）

给队列设置 TTL，过期后由死信交换机转投"取消"队列。但它有一个致命缺陷——队头阻塞：RabbitMQ 只检查队首消息是否过期，如果队首是 30 分钟的消息，后面 1 分钟就该过期的消息也得排队等它。

方案二：延迟消息插件（推荐）

安装 rabbitmq_delayed_message_exchange 插件后，使用 x-delayed-message 类型的交换机，延迟时间写在消息头里，彻底避免队头阻塞：

@Bean
public CustomExchange delayExchange() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "direct");
    return new CustomExchange("order.delay.exchange", "x-delayed-message", true, false, args);
}

/** 30 分钟后提醒取消订单 */
public void sendDelayCancel(String orderId, long delayMillis) {
    rabbitTemplate.convertAndSend("order.delay.exchange", "order.cancel", orderId, msg -> {
        msg.getMessageProperties().setDelay((int) delayMillis);   // 写入 x-delay 头
        return msg;
    });
}

消费端和普通队列一样用 @RabbitListener 监听即可，到期后消息才会投递过来。

顺带一提：RocketMQ 原生支持延迟消息（messageDelayLevel，5.x 支持任意时间），Kafka 则需要借助定时扫表或时间轮方案自建——这正是一些交易类团队选 RocketMQ 的原因。

八、Spring Kafka 速览

前面所有可靠性思路在 Kafka 中一一对应，只是配置不同：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all                       # 对应 RabbitMQ 的 Confirm：等待所有 ISR 副本写入
      properties:
        enable.idempotence: true      # 生产者幂等，避免重试导致的 Broker 端重复
    consumer:
      group-id: order-group
      auto-offset-reset: earliest
      enable-auto-commit: false       # 手动提交 offset，对应手动 ACK
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.example.mq.dto"
    listener:
      ack-mode: manual_immediate      # 业务处理成功后立即提交

@KafkaListener(topics = "order-events", groupId = "order-group")
public void onMessage(ConsumerRecord<String, OrderMessage> record, Acknowledgment ack) {
    log.info("partition={}, offset={}, value={}", record.partition(), record.offset(), record.value());
    orderService.handle(record.value());   // 同样要幂等！
    ack.acknowledge();                      // 处理成功才提交 offset
}

两个 Kafka 特有的注意点：

1. 反序列化失败会死循环：遇到脏数据，JsonDeserializer 反复失败导致消费者卡死，生产环境应包一层 ErrorHandlingDeserializer，失败消息交给 DeadLetterPublishingRecoverer 转发到死信 topic；
2. offset 提交与幂等：acknowledge() 提交的可能只是"处理完了"，Kafka 的重平衡仍可能导致重复消费，消费端幂等（第五节）同样是必备项。

九、最佳实践清单

生产环境踩过坑之后，我总结了这份清单，建议直接收藏对照：

1. 消费端永远做幂等——重复投递是常态，去重标记 + DB 唯一索引兜底；
2. 不要在数据库事务里发 MQ——用本地消息表（或 RocketMQ 事务消息），避免"业务回滚了、消息已发出"的幽灵消息；
3. 失败消息有限重试 + 死信落库，绝不无限 requeue，毒消息会拖垮整个队列；
4. 消息体只传业务键——大字段、文件放 OSS，消息里传 URL；消息体越大，吞吐越低；
5. 给消费者设置合理的 prefetch/QoS——太大压垮消费者、内存溢出，太小吞吐上不去；
6. 消息结构向前兼容——只加字段不删字段、不改字段含义，消费者对未知字段要容忍；
7. 监控三件套：队列堆积量、消费延迟（消息从投递到被消费的耗时）、死信数量告警——堆积往往是事故的第一信号；
8. 别用 MQ 做同步 RPC——你不知道对方什么时候消费，超时和重试语义都会失控；
9. 连接与信道：RabbitMQ 的 Connection 是 TCP 级资源，Spring AMQP 已做好连接复用与 Channel 池化，不要自己手动管理连接；
10. 为 MQ 集群做容量与高可用规划——RabbitMQ 用 Quorum 队列 + 集群，Kafka 调好副本因子（replication.factor >= 3）与 min.insync.replicas。

十、写在最后

回顾全文，Spring 集成消息队列的代码本身并不难——几个注解、几十行配置。真正决定系统质量的是那些"代码之外"的问题：

• 消息不丢：生产者确认 + 持久化 + 手动 ACK，三道防线缺一不可；
• 消息不重：放弃"投递不重复"的幻想，用幂等消费拥抱重复；
• 消息有序：分而治之，同业务 ID 串行、不同业务 ID 并行；
• 失败兜底：死信队列 + 落库补偿，让每条消息都有出路。

消息队列是架构从"单体同步"走向"分布式异步"的第一步，也是最重要的一步。理解了它的语义边界，后续的事件驱动架构（EDA）、CQRS、事件溯源，不过是同一套思想在不同规模上的延伸。

如果这篇文章对你有帮助，欢迎点赞、收藏、转发三连，评论区聊聊你们踩过的 MQ 的坑。

环境说明：本文代码基于 Spring Boot 3.x（Spring AMQP 3.x / Spring Kafka 3.x）、RabbitMQ 3.13、JDK 17。