您的位置: 首页> Java源码

以为发消息=下单成功？RabbitMQ从0到秒杀实战的完整踩坑笔记

匿名上传

发布时间:2026-02-27 16:00:02

大二实战踩坑实录：我把 RabbitMQ 从“能跑”改成“尽量不丢”，被线上教育后的完整笔记

前言：万恶之源起于“秒杀”

大家好，这是我写的第一篇博客！作为一名刚入门后端的大二学生，以前写单体项目时，遇到业务都是一套同步代码从头调到尾，主打一个“能跑就行”。

但直到我接手了项目里的**“秒杀下单”**模块，才体会到了什么叫“社会的毒打”。在秒杀场景下，瞬间涌入的成千上万个请求如果直接砸进 MySQL，数据库怕是当场就要“拔管”。为了保住可怜的数据库，我们要引入 RabbitMQ。

1. 到底为什么要用 RabbitMQ？

在刚学的时候，我总觉得这玩意儿把系统搞复杂了，但真正用到高并发场景才知道它有多香。总结起来就是经典的三板斧：

异步（提速）： 用户点击“抢购”，系统只要在 Redis 里扣减完库存，直接把订单信息扔给 RabbitMQ，然后立刻给用户返回“排队中”。剩下的生成订单、扣减积分、发短信等操作全在后台慢慢做，接口响应时间从几秒缩短到几十毫秒。
削峰填谷（保命）： 假如一秒钟来了 1万个下单请求，但我的 MySQL 每秒最多只能扛 1000 个。有了 MQ 之后，这 1万个请求变成消息堆积在 MQ 的队列里，消费端按照 1000/秒的速度慢慢拉取处理。请求高峰被完美“削平”，系统安然无恙。
解耦（防连带雪崩）： 如果是同步调用，一旦“发短信”服务宕机挂掉，整个下单流程都会卡死失败。引入 MQ 后，下单主流程只管发消息，根本不需要管短信服务死没死。短信服务恢复后，自己去队列里把积压的消息消费掉就行了。

2. RabbitMQ 的核心基础概念（大白话版）

没学之前看官方文档一头雾水，其实把它当成**“寄快递”**秒懂：

Producer（生产者）： 寄件人。也就是我们的发消息的代码，只负责把消息发出去。
Consumer（消费者）： 收件人。并在后台处理消息的代码。
Broker（消息服务器）： 快递中转站。RabbitMQ 的服务端程序本身。
Exchange（交换机）： 快递分拣中心。生产者发出的消息先到达这里，它不存消息，只负责根据规则把消息“分拣”到指定的队列。
Queue（队列）： 快递柜。真正存放消息的地方，消息会在这里排队，等待消费者来取。
RoutingKey（路由键）： 快递单上的包裹地址。生产者发消息时带上的“标签”。
Binding（绑定）： 分拣规则。规定了交换机如果看到某种 RoutingKey，就把它扔进哪个 Queue 里。

了解完这些高大上的概念，我以为我已经无敌了，我的目标特别朴素：消息发出去，消费者能收到，就算成功。

结果真上手秒杀场景后，现实直接给我一记重锤。

问题痛点

我一开始写完 MQ 逻辑，测试没问题就提交了，结果遇到了以下情况：

Redis 已经预扣库存了
用户下单标记也打上了
结果 MQ 消息因为各种网络抖动、代码异常没被正常消费
数据库里没订单，用户却再也下不了单（因为 Redis 里已经记成“已下单”）

这就很尴尬：
用户以为自己下单成功，我以为系统挺稳定，数据库表示“关我啥事”。

再加上这些常见的灵魂拷问：

消息到达 Broker 了吗？（生产者根本不知道）
消息能路由到队列吗？（交换机和路由键写错一位都寄，消息直接消失）
消费失败怎么办？直接丢弃还是无限重试？
重试也失败了怎么办？谁来收尸？

说白了，刚开始的我只做到了“能发能收”，离“生产可落地”还差一大截。

排查思路

我后面把问题拆成了 4 层来排查，每层都要有兜底：

生产者到 Broker：Confirm
- 确认消息是否被 RabbitMQ 收到
交换机到队列：Return
- 消息是否成功路由到目标队列
队列到消费者：手动 ACK + 有限重试
- 业务成功才 ACK，失败就重试，别一上来就进死信队列
重试失败后的最终一致性：DLQ + 补偿
- 消息彻底处理不动了，进死信队列（DLQ），做回滚补偿

我当时给自己立了个目标：

核心代码实现

下面是我项目里最关键的几段代码（按链路顺序放），注释我写得比较啰嗦，方便面试和复盘。

1）秒杀主入口：Redis 预检成功后投递 MQ，投递失败立刻补偿

@Override
public Result seckillVoucher(Long voucherId) {
    Long userId = UserHolder.getUser().getId();
    long orderId = redisIdWorker.nextId("order");

    // 1. Redis + Lua 原子校验：库存是否足够、是否重复下单
    Long result = stringRedisTemplate.execute(
            SECKILL_SCRIPT,
            Collections.emptyList(),
            voucherId.toString(), userId.toString()
    );

    int r = result.intValue();
    if (r != 0) {
        // r=1 库存不足；r=2 重复下单
        return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
    }

    // 2. 校验通过，构造订单消息体（这里只是消息，不直接落库）
    VoucherOrder voucherOrder = new VoucherOrder();
    voucherOrder.setId(orderId);
    voucherOrder.setUserId(userId);
    voucherOrder.setVoucherId(voucherId);

    // 3. 发送 MQ：让消费端异步落库，削峰填谷
    try {
        rabbitTemplate.convertAndSend(MqConfig.EXCHANGE, MqConfig.KEY, voucherOrder);
    } catch (Exception e) {
        // 4. 发送失败：Redis 里已经做了预扣和下单标记，必须补偿回滚
        log.error("秒杀下单消息发送失败，触发补偿回滚。voucherId={}, userId={}, orderId={}",
                voucherId, userId, orderId, e);
        seckillCompensationService.compensate(voucherId, userId, "MQ发送异常");
        return Result.fail("系统繁忙，请稍后重试");
    }

    // 5. 快速返回：用户先拿到“排队中”
    return Result.ok(orderId);
}

2）MQ 配置：业务队列绑定死信交换机 + 手动 ACK 容器

@Configuration
public class MqConfig {

    // 秒杀主链路
    public static final String EXCHANGE = "seckill.topic";
    public static final String QUEUE = "seckill.queue";
    public static final String KEY = "seckill.order";

    // 死信链路
    public static final String DLX_EXCHANGE = "seckill.dlx.topic";
    public static final String DLX_QUEUE = "seckill.dlx.queue";
    public static final String DLX_KEY = "seckill.dlx";

    @Bean
    public TopicExchange seckillExchange() {
        return new TopicExchange(EXCHANGE);
    }

    @Bean
    public Queue seckillQueue() {
        // 关键：业务队列绑定死信交换机
        // 当消息被 NACK 且 requeue=false 时，会进 DLQ
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", DLX_EXCHANGE);
        args.put("x-dead-letter-routing-key", DLX_KEY);
        return new Queue(QUEUE, true, false, false, args);
    }

    @Bean
    public TopicExchange seckillDlxExchange() {
        return new TopicExchange(DLX_EXCHANGE);
    }

    @Bean
    public Queue seckillDlxQueue() {
        return new Queue(DLX_QUEUE, true);
    }

    @Bean
    public Binding bindSeckillQueue() {
        return BindingBuilder.bind(seckillQueue()).to(seckillExchange()).with(KEY);
    }

    @Bean
    public Binding bindSeckillDlxQueue() {
        return BindingBuilder.bind(seckillDlxQueue()).to(seckillDlxExchange()).with(DLX_KEY);
    }

    @Bean
    public SimpleRabbitListenerContainerFactory manualAckContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 手动ACK：业务处理成功后再确认消息
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 每次只拉一条，避免未ACK消息堆在消费者内存里
        factory.setPrefetchCount(1);
        return factory;
    }
}

3）消费者：手动 ACK + 有限重试（最多 3 次）+ 超限进 DLQ

@RabbitListener(queues = MqConfig.QUEUE, containerFactory = "manualAckContainerFactory")
public void listenSeckillQueue(VoucherOrder voucherOrder, Channel channel, Message message) throws IOException {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();

    // 1. 用户维度加锁：避免并发重复处理同一用户订单
    RLock lock = redissonClient.getLock("lock:order:" + voucherOrder.getUserId());
    boolean isLock = lock.tryLock();
    if (!isLock) {
        // 当前有相同用户订单在处理，直接ACK丢弃当前消息，防止重复消费风暴
        channel.basicAck(deliveryTag, false);
        return;
    }

    try {
        // 2. 真正执行业务（事务方法）
        proxy.createVoucherOrder(voucherOrder);

        // 3. 成功后手动ACK
        channel.basicAck(deliveryTag, false);

    } catch (Exception e) {
        // 4. 读取自定义重试次数
        Integer retry = (Integer) message.getMessageProperties().getHeaders().getOrDefault("x-retry", 0);

        if (retry < 3) {
            int nextRetry = retry + 1;

            // 5. 重新投递一条新消息，并把 x-retry +1
            MessagePostProcessor mpp = m -> {
                m.getMessageProperties().setHeader("x-retry", nextRetry);
                return m;
            };
            rabbitTemplate.convertAndSend(MqConfig.EXCHANGE, MqConfig.KEY, voucherOrder, mpp);

            // 6. 重投成功后，ACK 当前失败消息，避免原消息反复立即重投
            channel.basicAck(deliveryTag, false);
        } else {
            // 7. 超过重试上限：NACK 且不重回队列，交给 DLQ 处理
            channel.basicNack(deliveryTag, false, false);
        }
    } finally {
        // 8. finally 解锁，防止死锁
        lock.unlock();
    }
}

4）生产者可靠性：Confirm + Return 双回调

@PostConstruct
public void initCallbacks() {

    // 1. ConfirmCallback：确认消息是否到达 Broker
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        String id = correlationData == null ? null : correlationData.getId();
        if (ack) {
            log.debug("MQ Confirm ACK, correlationId={}", id);
        } else {
            // 这里只能说明“Broker没收到”，通常要配合本地消息表/重发任务
            log.error("MQ Confirm NACK, correlationId={}, cause={}", id, cause);
        }
    });

    // 2. ReturnCallback：消息到达交换机但无法路由到队列
    rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
        try {
            String body = new String(message.getBody());
            log.error("MQ Return(不可路由): replyCode={}, replyText={}, exchange={}, routingKey={}, body={}",
                    replyCode, replyText, exchange, routingKey, body);

            // 解析消息，触发补偿，避免Redis状态和DB状态“各走各的”
            Map<String, Object> map = objectMapper.readValue(body, Map.class);
            Long voucherId = map.get("voucherId") == null ? null : Long.valueOf(map.get("voucherId").toString());
            Long userId = map.get("userId") == null ? null : Long.valueOf(map.get("userId").toString());
            seckillCompensationService.compensate(voucherId, userId, "消息不可路由(Return)");
        } catch (Exception e) {
            log.error("处理 MQ Return 时异常", e);
        }
    });
}

5）死信消费 + 补偿服务：最终一致性兜底

@RabbitListener(queues = MqConfig.DLX_QUEUE, containerFactory = "manualAckContainerFactory")
public void listenSeckillDlx(Message message, Channel channel) throws IOException {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    try {
        String body = new String(message.getBody());
        VoucherOrder voucherOrder = objectMapper.readValue(body, VoucherOrder.class);

        Long orderId = voucherOrder.getId();
        Long voucherId = voucherOrder.getVoucherId();
        Long userId = voucherOrder.getUserId();

        // 1. 如果DB已经有订单，说明只是消息链路问题，不需要补偿
        boolean orderExists = voucherOrderService.query().eq("id", orderId).count() > 0;

        // 2. 如果DB没有订单，执行补偿：回滚Redis标记、校准库存
        if (!orderExists) {
            seckillCompensationService.compensate(voucherId, userId, "消费失败进入DLQ");
        }

        channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
        // 3. DLQ处理再失败也别无限循环，直接ACK并记录日志
        log.error("处理秒杀死信消息异常，直接确认丢弃", e);
        channel.basicAck(deliveryTag, false);
    }
}

public void compensate(Long voucherId, Long userId, String reason) {
    if (voucherId == null || userId == null) {
        log.error("补偿参数非法：voucherId={}, userId={}, reason={}", voucherId, userId, reason);
        return;
    }

    // 1. 回滚“已下单”标记，让用户可再次尝试
    stringRedisTemplate.opsForSet().remove("seckill:order:" + voucherId, userId.toString());

    // 2. 按数据库库存校准 Redis 库存（以DB为准）
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    int dbStock = (voucher == null || voucher.getStock() == null) ? 0 : voucher.getStock();
    stringRedisTemplate.opsForValue().set("seckill:stock:" + voucherId, String.valueOf(dbStock));

    log.warn("秒杀补偿完成：voucherId={}, userId={}, dbStock={}, reason={}", voucherId, userId, dbStock, reason);
}