10 个MQ高频业务场景深度解析

在分布式系统开发中，很多开发者都会疑惑：“明明直接调用接口就能实现功能，为什么还要引入MQ这个‘中间商’？”其实，当系统面临高并发、紧耦合、数据一致性等问题时，MQ的价值才会真正凸显。本文结合实际业务场景，从问题背景、解决方案、代码实现到技术要点，全面拆解MQ的10种核心用法，帮你避开常见陷阱，真正用好消息队列。

一、MQ的核心价值：为什么需要它？

在引入具体场景前，先明确MQ解决的核心问题。系统间直接调用会面临四大痛点：

• 紧耦合：上游服务需感知所有下游服务，新增或移除下游需修改上游代码。
• 性能瓶颈：同步调用会累积所有下游服务的响应时间，导致接口耗时过长。
• 单点故障：任一下游服务宕机，会导致上游服务调用失败，影响核心流程。
• 扩展困难：下游服务扩容时，上游服务可能需要同步调整调用逻辑。

而引入MQ后，能实现五大核心能力：

• 解耦：上游只需发送消息，无需关心下游如何处理。
• 异步化：非核心流程异步执行，减少主流程响应时间。
• 削峰填谷：缓冲突发流量，避免数据库或服务被瞬时请求压垮。
• 冗余存储：消息持久化，避免数据丢失。
• 一致性保障：通过事务消息、重试机制，保证分布式系统数据一致性。

二、10个实战场景：从问题到解决方案

场景一：系统解耦——打破服务间强依赖

背景痛点

电商项目中，订单创建后需同步调用库存、积分、邮件、数据分析等多个服务（代码如下）。这种设计下，新增一个下游服务（如物流通知）需修改订单服务代码，且任一下游服务故障（如邮件服务超时）会导致订单创建失败。

// 紧耦合代码示例
@Service
public class OrderService {
    // 依赖多个下游服务
    @Autowired private InventoryService inventoryService;
    @Autowired private PointsService pointsService;
    @Autowired private EmailService emailService;

    public void createOrder(Order order) {
        // 1. 保存订单（核心流程）
        orderDao.save(order);
        // 2. 同步调用库存服务（非核心，但故障会阻塞订单）
        inventoryService.updateInventory(order);
        // 3. 同步调用积分服务（非核心）
        pointsService.addPoints(order.getUserId(), order.getAmount());
        // 4. 同步发送邮件（非核心，超时风险高）
        emailService.sendConfirmation(order);
    }
}

MQ解决方案

引入MQ后，订单服务只需发送“订单创建”消息到MQ，下游服务各自订阅消息并处理，服务间彻底解耦。架构如下：

订单服务 → 发送消息 → MQ → 库存服务/积分服务/邮件服务（各自消费）

代码实现

// 1. 订单服务（生产者）：只关注核心流程，发送消息
@Service
public class OrderService {
    @Autowired private RabbitTemplate rabbitTemplate;

    public void createOrder(Order order) {
        // 核心：保存订单
        orderDao.save(order);
        // 发送消息到MQ，参数：交换机、路由键、消息体
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(), order.getUserId(), order.getAmount()
        );
        rabbitTemplate.convertAndSend(
            "order.exchange",  // 交换机名称
            "order.created",   // 路由键（匹配下游队列）
            event
        );
    }
}

// 2. 库存服务（消费者）：订阅消息，独立处理
@Component
public class InventoryConsumer {
    @Autowired private InventoryService inventoryService;

    // 监听指定队列
    @RabbitListener(queues = "inventory.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 独立处理库存更新，失败不影响订单服务
        inventoryService.updateInventory(event.getOrderId());
    }
}

// 3. 邮件服务（消费者）：同理订阅消息
@Component
public class EmailConsumer {
    @Autowired private EmailService emailService;

    @RabbitListener(queues = "email.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        emailService.sendConfirmation(event.getUserId(), event.getOrderId());
    }
}

技术要点

• 交换机与队列绑定：通过路由键（如order.created）将交换机与下游队列绑定，实现消息精准投递。
• 错误处理：下游服务失败时，通过重试机制（如RabbitMQ的重新入队）或死信队列兜底，避免消息丢失。
• 消息格式：使用JSON或Protobuf定义消息结构，保证跨服务、跨语言兼容性。

场景二：异步处理——提升核心流程响应速度

背景痛点

用户上传视频后，需执行转码、生成缩略图、内容审核等耗时操作（单步可能需几秒）。若同步处理，用户需等待操作完成才能收到反馈，体验极差。

MQ解决方案

视频服务接收上传后，立即保存原始视频并返回结果，同时发送“视频处理”消息到MQ；后台消费者异步执行转码、审核等操作，处理完成后通知用户。流程如下：

用户 → 视频服务（保存视频+发送消息）→ 返回上传成功 → MQ → 处理服务（转码/审核）→ 通知用户

代码实现

// 1. 视频服务（生产者）：快速响应用户
@Service
public class VideoService {
    @Autowired private KafkaTemplate<String, Object> kafkaTemplate;

    public UploadResponse uploadVideo(MultipartFile file, String userId) {
        // 1. 保存原始视频，生成唯一视频ID
        String videoId = saveOriginalVideo(file);
        // 2. 发送处理消息到Kafka（异步执行，不阻塞返回）
        VideoProcessingEvent event = new VideoProcessingEvent(videoId, userId);
        kafkaTemplate.send("video-processing-topic", event);
        // 3. 立即返回，用户无需等待
        return new UploadResponse(videoId, "上传成功，处理中");
    }
}

// 2. 视频处理服务（消费者）：后台异步处理
@Service
public class VideoProcessingConsumer {
    @Autowired private VideoProcessor videoProcessor;
    @Autowired private NotificationService notificationService;

    // 监听Kafka主题
    @KafkaListener(topics = "video-processing-topic")
    public void processVideo(VideoProcessingEvent event) {
        try {
            // 耗时操作：转码、生成缩略图、内容审核
            videoProcessor.transcode(event.getVideoId());
            videoProcessor.generateThumbnails(event.getVideoId());
            contentAuditService.check(event.getVideoId());
            // 处理完成，通知用户
            notificationService.notifyUser(event.getUserId(), "视频处理完成");
        } catch (Exception e) {
            log.error("视频处理失败，videoId:{}", event.getVideoId(), e);
            // 失败重试：可发送到重试队列
        }
    }
}

技术要点

• MQ选型：处理大量视频消息时，优先选Kafka（高吞吐、持久化能力强）。
• 响应设计：返回给用户“处理中”状态，并提供查询接口（如通过videoId查询进度）。
• 弹性扩展：若处理压力大，可增加消费者实例，MQ自动负载均衡消息。

场景三：流量削峰——应对秒杀等突发流量

背景痛点

电商秒杀活动中，瞬时请求量可能达到平时的100倍（如1秒10万请求）。若直接转发到数据库，会导致数据库连接耗尽、查询超时，甚至宕机。

MQ解决方案

通过MQ缓冲请求，将“瞬时高并发”转化为“匀速消费”：

1. 网关层限流：拒绝超出承载能力的请求（如每秒只允许1万请求进入）。
2. MQ缓冲：通过队列存储秒杀请求，避免直接冲击数据库。
3. 消费者匀速处理：控制消费者数量，确保数据库按承载能力处理订单（如每秒处理500单）。

代码实现

// 1. 秒杀服务（生产者）：预减库存+发送消息
@Service
public class SecKillService {
    @Autowired private RedisTemplate<String, Object> redisTemplate;
    @Autowired private RabbitTemplate rabbitTemplate;

    public SecKillResponse secKill(SecKillRequest request) {
        String userId = request.getUserId();
        String itemId = request.getItemId();

        // 1. 校验用户资格（如是否已秒杀过）
        if (Boolean.TRUE.equals(redisTemplate.hasKey("sec_kill:user:" + userId + ":" + itemId))) {
            return SecKillResponse.failed("已参与秒杀");
        }

        // 2. Redis预减库存（原子操作，避免超卖）
        String stockKey = "sec_kill:stock:" + itemId;
        Long remainingStock = redisTemplate.opsForValue().decrement(stockKey);
        if (remainingStock == null || remainingStock < 0) {
            // 库存不足，恢复Redis计数
            redisTemplate.opsForValue().increment(stockKey);
            return SecKillResponse.failed("库存已空");
        }

        // 3. 发送秒杀成功消息到MQ，异步创建订单
        SecKillSuccessEvent event = new SecKillSuccessEvent(userId, itemId);
        rabbitTemplate.convertAndSend("sec_kill.exchange", "sec_kill.success", event);

        // 4. 标记用户已参与秒杀
        redisTemplate.opsForValue().set("sec_kill:user:" + userId + ":" + itemId, "1", 24, TimeUnit.HOURS);
        return SecKillResponse.success("秒杀成功，订单创建中");
    }
}

// 2. 订单处理服务（消费者）：匀速创建订单
@Component
public class SecKillOrderConsumer {
    @Autowired private OrderService orderService;

    // 单消费者实例控制处理速度，或通过线程池调节
    @RabbitListener(queues = "sec_kill.order.queue")
    public void createSecKillOrder(SecKillSuccessEvent event) {
        try {
            // 匀速处理：创建订单（数据库操作）
            orderService.createSecKillOrder(event.getUserId(), event.getItemId());
        } catch (Exception e) {
            log.error("创建秒杀订单失败，userId:{}, itemId:{}", event.getUserId(), event.getItemId(), e);
            // 失败处理：恢复库存+通知用户
            restoreStock(event.getItemId());
            notificationService.notifyUser(event.getUserId(), "秒杀订单创建失败");
        }
    }

    // 恢复库存（补偿逻辑）
    private void restoreStock(String itemId) {
        redisTemplate.opsForValue().increment("sec_kill:stock:" + itemId);
    }
}

技术要点

• 超卖防护：用Redis原子操作（decrement）预减库存，避免多线程下超卖。
• 限流控制：网关层用Nginx或Sentinel限流，避免MQ被过多请求压满。
• 库存一致性：秒杀结束后，需对比Redis库存与数据库库存，修正差异（如Redis漏减）。

场景四：数据同步——保证微服务数据一致性

背景痛点

微服务架构中，每个服务有独立数据库（如用户服务用user_db，订单服务用order_db）。当用户状态变更（如禁用账号）时，订单服务的本地用户缓存需同步更新，否则会出现“用户已禁用但仍能下单”的问题。

MQ解决方案

用户服务数据变更时，发送“用户更新”消息到MQ；订单、支付等依赖用户数据的服务，订阅消息并更新本地缓存或数据库，实现数据最终一致性。

代码实现

// 1. 用户服务（生产者）：事务内发送消息，保证数据与消息一致
@Service
public class UserService {
    @Autowired private RocketMQTemplate rocketMQTemplate;

    // 本地事务：更新数据库+发送消息
    @Transactional
    public void updateUserStatus(String userId, String status) {
        // 1. 更新用户状态（数据库事务）
        User user = userDao.findById(userId).orElseThrow(() -> new RuntimeException("用户不存在"));
        user.setStatus(status);
        userDao.save(user);

        // 2. 发送事务消息（RocketMQ支持，确保消息与数据库操作同成功/同失败）
        UserStatusEvent event = new UserStatusEvent(userId, status);
        rocketMQTemplate.sendMessageInTransaction(
            "user-status-topic",  // 主题
            MessageBuilder.withPayload(event).build(),
            null  // 事务参数（可选）
        );
    }
}

// 2. 订单服务（消费者）：同步更新本地缓存
@Component
@RocketMQMessageListener(
    topic = "user-status-topic",
    consumerGroup = "order-service-group"  // 消费者组，同一组内负载均衡
)
public class UserStatusConsumer implements RocketMQListener<UserStatusEvent> {
    @Autowired private LocalCacheManager localCacheManager;

    @Override
    public void onMessage(UserStatusEvent event) {
        // 更新本地用户缓存（如禁用用户，后续下单会校验）
        localCacheManager.updateUserStatus(event.getUserId(), event.getStatus());
        // 可选：标记用户相关订单为“不可操作”
        orderService.markOrderByUserStatus(event.getUserId(), event.getStatus());
    }
}

技术要点

• 事务消息：用RocketMQ的事务消息机制，确保“数据库更新成功”与“消息发送成功”原子性（避免数据更新但消息未发，或消息发了但数据未更）。
• 幂等消费：消费者需实现幂等（如通过消息ID去重），避免重复更新缓存。
• 延迟处理：若下游服务依赖数据未准备好，可通过延迟队列重试（如3秒后再更新缓存）。

场景五：日志收集——分布式日志集中处理

背景痛点

分布式系统中，日志分散在多台服务器的本地文件中（如应用节点1的日志在/var/log/app1/，节点2在/var/log/app2/），排查问题时需逐个登录服务器查看，效率极低。

MQ解决方案

通过MQ实现日志“集中收集-分发处理”：

1. 应用节点：将日志发送到MQ（如Kafka）。
2. 消费服务：从MQ获取日志，分别写入Elasticsearch（用于查询）、HDFS（用于归档）、监控系统（用于告警）。

代码实现

// 1. 日志收集组件（生产者）：嵌入应用，发送日志
@Component
public class LogCollector {
    @Autowired private KafkaTemplate<String, String> kafkaTemplate;

    // 收集日志并发送到Kafka
    public void collect(String appId, String level, String message, Map<String, Object> context) {
        // 构建日志对象
        LogEvent logEvent = new LogEvent();
        logEvent.setAppId(appId);
        logEvent.setLevel(level);
        logEvent.setMessage(message);
        logEvent.setContext(context);
        logEvent.setTimestamp(System.currentTimeMillis());

        // 发送到Kafka，按appId分区（便于按应用筛选）
        kafkaTemplate.send(
            "app-logs-topic",
            appId,  // 分区键：同一应用的日志进入同一分区，保证顺序
            JsonUtils.toJson(logEvent)
        );
    }
}

// 2. 日志处理服务（消费者1）：写入Elasticsearch，支持查询
@Service
public class EsLogConsumer {
    @Autowired private ElasticsearchRestTemplate esTemplate;

    @KafkaListener(topics = "app-logs-topic", groupId = "log-es-group")
    public void writeToEs(String logJson) {
        LogEvent logEvent = JsonUtils.fromJson(logJson, LogEvent.class);
        // 写入Elasticsearch，索引按日期拆分（如app-logs-20251017）
        String index = "app-logs-" + DateUtils.format(new Date(), "yyyyMMdd");
        esTemplate.save(logEvent, index);
    }
}

// 3. 日志告警服务（消费者2）：异常日志触发告警
@Service
public class AlertLogConsumer {
    @Autowired private AlertService alertService;

    @KafkaListener(topics = "app-logs-topic", groupId = "log-alert-group")
    public void checkAndAlert(String logJson) {
        LogEvent logEvent = JsonUtils.fromJson(logJson, LogEvent.class);
        //  ERROR级日志触发告警
        if ("ERROR".equals(logEvent.getLevel())) {
            AlertMessage alert = new AlertMessage();
            alert.setTitle("应用异常日志");
            alert.setContent("appId:" + logEvent.getAppId() + ", 消息:" + logEvent.getMessage());
            alertService.sendAlert(alert);  // 发送邮件/短信告警
        }
    }
}

技术要点

• MQ选型：优先选Kafka（高吞吐，支持海量日志收集）。
• 日志分区：按应用ID或服务名分区，避免单分区日志过多导致处理缓慢。
• 压缩传输：开启Kafka消息压缩（如GZIP），减少网络带宽消耗。

场景六：消息广播——配置更新实时同步

背景痛点

系统配置（如限流阈值、功能开关）更新后，需通知所有服务节点（如10台应用服务器）更新本地缓存，否则部分节点仍使用旧配置，导致行为不一致。

MQ解决方案

配置服务更新配置后，发送“配置更新”广播消息；所有服务节点订阅消息，实时更新本地配置缓存。常用MQ：RabbitMQ的Fanout交换机（广播）或Redis的Pub/Sub（轻量级广播）。

代码实现（Redis Pub/Sub）

// 1. 配置服务（发布者）：广播配置更新
@Service
public class ConfigService {
    @Autowired private RedisTemplate<String, Object> redisTemplate;

    public void updateConfig(String configKey, String configValue) {
        // 1. 保存配置到数据库/Redis
        configDao.updateConfig(configKey, configValue);
        // 2. 广播配置更新消息（Redis Pub/Sub）
        ConfigUpdateEvent event = new ConfigUpdateEvent(configKey, configValue);
        redisTemplate.convertAndSend("config-update-channel", event);
    }
}

// 2. 应用节点（订阅者）：接收消息并更新本地缓存
@Component
public class ConfigSubscriber {
    @Autowired private LocalConfigCache localConfigCache;

    // 订阅Redis频道
    @RedisMessageListener(channel = "config-update-channel")
    public void onConfigUpdate(ConfigUpdateEvent event) {
        // 更新本地配置缓存（内存缓存，如Caffeine）
        localConfigCache.update(event.getConfigKey(), event.getConfigValue());
        log.info("配置更新：{}={}", event.getConfigKey(), event.getConfigValue());
    }
}

技术要点

• 可靠性选择：Redis Pub/Sub不保证消息持久化（节点离线会丢消息），若需可靠广播，用RabbitMQ的Fanout交换机+持久化队列。
• 配置版本：消息中携带配置版本号，避免旧消息覆盖新配置（如节点离线后收到旧消息）。

场景七：顺序消息——保证业务流程有序性

背景痛点

订单状态变更需按“创建→支付→发货→完成”的顺序处理，若消息乱序（如“发货”消息先于“支付”消息到达），会导致业务逻辑错误（如未支付就发货）。

MQ解决方案

通过“分区+顺序消费”保证消息有序：

1. 发送端：同一订单的消息，按订单ID哈希到同一MQ分区（如Kafka的分区、RocketMQ的队列），确保同一订单的消息在同一分区内有序。
2. 消费端：每个分区用单线程消费，避免多线程乱序。

代码实现（RocketMQ）

// 1. 订单服务（生产者）：按订单ID分区，保证同一订单消息有序
@Service
public class OrderStateService {
    @Autowired private RocketMQTemplate rocketMQTemplate;

    public void changeOrderState(String orderId, String oldState, String newState) {
        OrderStateEvent event = new OrderStateEvent(orderId, oldState, newState);
        // 发送顺序消息：第三个参数为shardingKey（订单ID），同一ID进入同一队列
        rocketMQTemplate.syncSendOrderly(
            "order-state-topic",  // 主题
            event,
            orderId  // 分片键：保证同一订单消息有序
        );
    }
}

// 2. 订单处理服务（消费者）：顺序消费
@Service
@RocketMQMessageListener(
    topic = "order-state-topic",
    consumerGroup = "order-state-group",
    consumeMode = ConsumeMode.ORDERLY  // 顺序消费模式（单线程处理每个队列）
)
public class OrderStateConsumer implements RocketMQListener<OrderStateEvent> {
    @Autowired private OrderService orderService;

    @Override
    public void onMessage(OrderStateEvent event) {
        // 按顺序处理订单状态变更（如“支付”完成后才处理“发货”）
        orderService.processStateChange(event.getOrderId(), event.getOldState(), event.getNewState());
    }
}

技术要点

• 分区数量：分区数需合理（如按订单量设置10-20个分区），避免单分区消息堆积。
• 故障处理：若某分区消费失败，需暂停该分区消费（避免阻塞其他分区），排查问题后重试。

场景八：延迟消息——实现定时任务

背景痛点

订单超时未支付需自动取消（如30分钟后），传统方案如“数据库轮询”会导致数据库压力大、实时性差（如每分钟轮询一次，最多延迟1分钟）。

MQ解决方案

使用MQ的延迟队列：订单创建时发送“延迟30分钟”的消息，消费者在30分钟后收到消息，检查订单支付状态，未支付则取消。

代码实现（RabbitMQ）

// 1. 订单服务（生产者）：发送延迟消息
@Service
public class OrderService {
    @Autowired private RabbitTemplate rabbitTemplate;

    public void createOrder(Order order) {
        // 1. 保存订单（状态：未支付）
        order.setStatus("UNPAID");
        orderDao.save(order);

        // 2. 发送延迟30分钟的消息
        OrderCreateEvent event = new OrderCreateEvent(order.getId());
        rabbitTemplate.convertAndSend(
            "order.delay.exchange",  // 延迟交换机
            "order.delay.routing",   // 延迟路由键
            event,
            message -> {
                // 设置延迟时间：30分钟（单位：毫秒）
                message.getMessageProperties().setDelay(30 * 60 * 1000);
                return message;
            }
        );
    }
}

// 2. 订单超时处理服务（消费者）：30分钟后检查
@Component
public class OrderTimeoutConsumer {
    @Autowired private OrderService orderService;

    @RabbitListener(queues = "order.delay.queue")
    public void checkOrderPayment(OrderCreateEvent event) {
        // 查询订单当前状态
        Order order = orderDao.findById(event.getOrderId()).orElse(null);
        if (order != null && "UNPAID".equals(order.getStatus())) {
            // 超时未支付，取消订单
            orderService.cancelOrder(order.getId(), "超时未支付");
            // 恢复库存
            inventoryService.restoreStock(order.getItems());
        }
    }
}

技术要点

• 延迟精度：RabbitMQ的延迟队列精度约为秒级，若需更高精度（如毫秒级），可使用RocketMQ的定时消息。
• 消息持久化：开启队列持久化，避免MQ重启后延迟消息丢失。

• 替代方案对比：	方案	优点	缺点
  数据库轮询	实现简单	数据库压力大，实时性差
  MQ延迟队列	实时性好，无轮询压力	依赖MQ，需处理消息堆积
  定时任务（如XXL-Job）	可控性强	分布式协调复杂，适合批量任务

场景九：消息重试——保证临时故障可恢复

背景痛点

消费者处理消息时，可能遇到临时故障（如数据库连接超时、下游服务临时不可用），若直接丢弃消息，会导致业务数据丢失。

MQ解决方案

通过“重试机制+死信队列”处理：

1. 临时故障：消息重新入队，重试几次（如3次）。
2. 永久故障：重试失败后，消息进入死信队列（DLQ），后续人工排查处理。

代码实现（RabbitMQ）

@Service
@Slf4j
public class RetryableConsumer {
    @Autowired private BusinessService businessService;

    // 监听业务队列，手动ACK（控制消息确认）
    @RabbitListener(queues = "business.queue")
    public void processMessage(Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 获取重试次数（自定义消息头）
        Integer retryCount = message.getMessageProperties().getHeader("retry_count");
        if (retryCount == null) retryCount = 0;

        try {
            // 1. 解析消息
            String msgBody = new String(message.getBody(), StandardCharsets.UTF_8);
            BusinessEvent event = JsonUtils.fromJson(msgBody, BusinessEvent.class);

            // 2. 业务处理
            businessService.process(event);

            // 3. 处理成功，手动ACK（消息从队列删除）
            channel.basicAck(deliveryTag, false);

        } catch (TemporaryException e) {
            // 临时故障（如数据库超时），重试（最多3次）
            if (retryCount < 3) {
                log.warn("临时故障，重试第{}次，消息:{}", retryCount + 1, msgBody, e);
                // 设置重试次数，重新入队
                message.getMessageProperties().setHeader("retry_count", retryCount + 1);
                channel.basicNack(deliveryTag, false, true);  // requeue=true：重新入队
            } else {
                // 重试超过3次，进入死信队列
                log.error("重试3次失败，进入死信队列，消息:{}", msgBody, e);
                channel.basicNack(deliveryTag, false, false);  // requeue=false：不重新入队
            }

        } catch (PermanentException e) {
            // 永久故障（如消息格式错误），直接进入死信队列
            log.error("永久故障，进入死信队列，消息:{}", msgBody, e);
            channel.basicNack(deliveryTag, false, false);
        }
    }
}

技术要点

• 手动ACK：关闭自动ACK（acknowledge-mode: MANUAL），确保业务处理成功后再确认消息。
• 重试间隔：可通过“延迟队列+重试次数”实现渐进式重试（如第1次间隔1秒，第2次3秒，第3次5秒）。
• 死信队列处理：定期监控死信队列，分析失败原因（如消息格式错误、业务逻辑bug），处理后可重新投递到业务队列。

场景十：事务消息——解决分布式事务问题

背景痛点

分布式系统中，“创建订单”和“扣减库存”需保证原子性（要么都成功，要么都失败）。若直接调用：

1. 先创建订单，再扣减库存：库存扣减失败，订单已创建，导致超卖。
2. 先扣减库存，再创建订单：订单创建失败，库存未恢复，导致库存锁定。

MQ解决方案

使用RocketMQ的事务消息，实现“本地事务+消息发送”的原子性：

1. 发送半消息：消息进入MQ，但标记为“不可消费”。
2. 执行本地事务：创建订单（数据库操作）。
3. 确认事务：本地事务成功则提交消息（下游可消费），失败则回滚消息（消息删除）。
4. 事务回查：若MQ未收到确认，定期回查本地事务状态，确保最终一致性。

代码实现

// 1. 订单服务（生产者）：发送事务消息
@Service
public class OrderTransactionService {
    @Autowired private RocketMQTemplate rocketMQTemplate;
    @Autowired private OrderDao orderDao;

    // 事务方法：创建订单+发送消息
    @Transactional
    public void createOrderWithTransaction(Order order) {
        // 1. 发送半消息（不可消费）
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
            "order-tx-topic",  // 事务消息主题
            MessageBuilder.withPayload(new OrderCreatedEvent(order.getId()))
                .setHeader("orderId", order.getId())  // 携带订单ID，用于回查
                .build(),
            order  // 事务参数（传递给本地事务方法）
        );

        // 2. 检查半消息发送结果
        if (!result.getLocalTransactionState().equals(LocalTransactionState.UNKNOWN)) {
            throw new RuntimeException("事务消息发送失败，订单创建中止");
        }
    }

    // 2. 本地事务执行器（RocketMQ回调）
    @RocketMQTransactionListener
    public class OrderTransactionListener implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {
                // arg为事务参数（订单对象）
                Order order = (Order) arg;
                // 执行本地事务：创建订单
                orderDao.save(order);
                // 本地事务成功，提交消息（下游可消费）
                return RocketMQLocalTransactionState.COMMIT_MESSAGE;
            } catch (Exception e) {
                // 本地事务失败，回滚消息（消息删除）
                return RocketMQLocalTransactionState.ROLLBACK_MESSAGE;
            }
        }

        // 3. 事务回查（MQ未收到确认时调用）
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // 从消息头获取订单ID
            String orderId = msg.getHeaders().get("orderId", String.class);
            // 回查本地事务状态：订单是否存在
            Order order = orderDao.findById(orderId).orElse(null);
            if (order != null) {
                // 订单存在，提交消息
                return RocketMQLocalTransactionState.COMMIT_MESSAGE;
            } else {
                // 订单不存在，回滚消息
                return RocketMQLocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
    }
}

// 4. 库存服务（消费者）：消费事务消息，扣减库存
@Component
@RocketMQMessageListener(
    topic = "order-tx-topic",
    consumerGroup = "inventory-service-group"
)
public class InventoryTxConsumer implements RocketMQListener<OrderCreatedEvent> {
    @Autowired private InventoryService inventoryService;

    @Override
    public void onMessage(OrderCreatedEvent event) {
        // 扣减库存（需实现幂等，避免重复扣减）
        inventoryService.deductStock(event.getOrderId());
    }
}

技术要点

• 幂等消费：库存扣减需通过“订单ID”去重（如扣减前检查是否已处理），避免MQ重试导致重复扣减。
• 事务回查：确保回查逻辑可靠（如通过订单ID查询数据库），避免因回查失败导致消息状态不确定。
• 补偿逻辑：若库存扣减失败，需触发补偿（如恢复订单状态、通知用户），或发送到死信队列人工处理。

三、MQ选型与最佳实践

1. MQ选型建议

2. 最佳实践

• 消息幂等：所有消费者必须实现幂等（如通过消息ID、业务唯一键去重），避免重复处理。
• 死信队列：为每个业务队列配置死信队列，兜底处理失败消息，避免消息丢失。
• 监控告警：监控MQ的消息堆积量、消费延迟、死信数量，超过阈值触发告警（如短信/邮件）。
• 性能优化：
  • 开启消息压缩（如Kafka的GZIP），减少网络传输。
  • 合理设置分区/队列数量，避免单分区堆积。
  • 消费者线程池与队列数量匹配，避免线程空闲或过载。

四、总结

MQ不是“银弹”，但在分布式系统中，它能有效解决解耦、异步、削峰、一致性等核心问题。关键在于根据业务场景选择合适的MQ和方案，同时做好消息可靠性、幂等性、监控告警等基础保障。

除非注明，否则均为李锋镝的博客原创文章，转载必须以链接形式标明本文链接

本文链接：https://www.lifengdi.com/zhong-jian-jian/4528