全部标签
写点什么
开源架构OpenHarmony算法元宇宙学习方法Web3.0高效工作数据库Python音视频前端AI大数据团队管理程序员运维深度思考低代码redisgolang微服务架构flutter 查看更多

消息重复消费 + 顺序性，分布式消息的终极难题？一线解决方案全解析！

作者：我爱娃哈哈😍
  • 2025-08-01
    湖北

  • 本文字数：5754 字

    阅读完需：约 19 分钟

一、开场白：消息重复消费，真能靠幂等性搞定？

还记得第一次遇到消息重复消费，老板一句话："你幂等性做了吗？"我一脸懵："幂等性？不就是重复消费吗？"结果一上线，要么订单重复创建，要么库存重复扣减，要么数据不一致！


今天咱们就聊聊，消息重复消费和顺序性保障到底怎么解决？为什么有的方案有效，有的方案无效？一线后端工程师的深度技术解析！


二、消息重复消费原理，先搞明白再解决

什么是消息重复消费？


  • 消息重复消费：同一条消息被消费者处理多次，导致业务逻辑重复执行。

  • 核心问题：数据不一致、业务逻辑错误、资源浪费。

  • 常见原因：网络重传、消费者重启、消息队列重试机制等。


为什么会出现重复消费？


  • 网络问题：网络抖动导致消息重传。

  • 消费者重启：消费者重启后重新消费消息。

  • 消息队列重试：消息处理失败时，队列自动重试。

  • 集群切换：主从切换时，消息可能重复发送。


三、幂等性解决方案

1. 基于数据库的幂等性

@Servicepublic class OrderService {        @Autowired    private OrderMapper orderMapper;        @Transactional    public void createOrder(OrderMessage message) {        // 检查订单是否已存在        Order existingOrder = orderMapper.selectByOrderId(message.getOrderId());        if (existingOrder != null) {            log.info("订单已存在，跳过处理: {}", message.getOrderId());            return;        }                // 创建订单        Order order = new Order();        order.setOrderId(message.getOrderId());        order.setUserId(message.getUserId());        order.setAmount(message.getAmount());        order.setStatus("CREATED");        order.setCreateTime(new Date());                orderMapper.insert(order);        log.info("订单创建成功: {}", message.getOrderId());    }}
复制代码

2. 基于 Redis 的幂等性

@Servicepublic class PaymentService {        @Autowired    private RedisTemplate<String, String> redisTemplate;        @Autowired    private PaymentMapper paymentMapper;        public void processPayment(PaymentMessage message) {        String key = "payment:" + message.getPaymentId();                // 使用Redis的SETNX命令实现幂等性        Boolean success = redisTemplate.opsForValue().setIfAbsent(key, "PROCESSING", 30, TimeUnit.MINUTES);                if (!success) {            log.info("支付已处理，跳过: {}", message.getPaymentId());            return;        }                try {            // 检查支付是否已存在            Payment existingPayment = paymentMapper.selectByPaymentId(message.getPaymentId());            if (existingPayment != null) {                log.info("支付已存在，跳过处理: {}", message.getPaymentId());                return;            }                        // 处理支付            Payment payment = new Payment();            payment.setPaymentId(message.getPaymentId());            payment.setOrderId(message.getOrderId());            payment.setAmount(message.getAmount());            payment.setStatus("SUCCESS");            payment.setCreateTime(new Date());                        paymentMapper.insert(payment);            log.info("支付处理成功: {}", message.getPaymentId());                    } finally {            // 删除Redis中的标记            redisTemplate.delete(key);        }    }}
复制代码

3. 基于消息 ID 的幂等性

@Componentpublic class MessageIdempotentHandler {        @Autowired    private RedisTemplate<String, String> redisTemplate;        public boolean isProcessed(String messageId) {        String key = "message:" + messageId;        return redisTemplate.hasKey(key);    }        public void markProcessed(String messageId) {        String key = "message:" + messageId;        redisTemplate.opsForValue().set(key, "PROCESSED", 24, TimeUnit.HOURS);    }        public boolean processMessage(String messageId, Runnable processor) {        if (isProcessed(messageId)) {            log.info("消息已处理，跳过: {}", messageId);            return false;        }                try {            processor.run();            markProcessed(messageId);            return true;        } catch (Exception e) {            log.error("消息处理失败: {}", e.getMessage());            return false;        }    }}
复制代码


四、消息顺序性保障方案

1. 单分区顺序消费

@Componentpublic class OrderConsumer {        @Autowired    private OrderService orderService;        @KafkaListener(        topics = "order_topic",        groupId = "order_group",        containerFactory = "kafkaListenerContainerFactory"    )    public void consumeOrder(String message) {        OrderMessage orderMessage = JSON.parseObject(message, OrderMessage.class);                // 确保同一用户订单按顺序处理        String partitionKey = orderMessage.getUserId();                synchronized (partitionKey.intern()) {            orderService.processOrder(orderMessage);        }    }}
复制代码

2. 基于消息 ID 的顺序性

@Servicepublic class SequentialMessageProcessor {        @Autowired    private RedisTemplate<String, String> redisTemplate;        public void processSequentialMessage(String userId, String messageId, Runnable processor) {        String key = "sequence:" + userId;                // 获取当前处理的消息ID        String currentMessageId = redisTemplate.opsForValue().get(key);                if (currentMessageId == null) {            // 第一条消息            redisTemplate.opsForValue().set(key, messageId);            processor.run();        } else {            // 检查消息顺序            if (isSequential(currentMessageId, messageId)) {                redisTemplate.opsForValue().set(key, messageId);                processor.run();            } else {                log.warn("消息顺序错误，跳过: {}", messageId);            }        }    }        private boolean isSequential(String currentId, String newId) {        // 实现消息顺序检查逻辑        return true;    }}
复制代码

3. 基于时间戳的顺序性

@Componentpublic class TimeBasedSequentialConsumer {        @Autowired    private OrderService orderService;        private final Map<String, Long> lastProcessedTime = new ConcurrentHashMap<>();        @KafkaListener(topics = "order_topic", groupId = "order_group")    public void consumeOrder(String message) {        OrderMessage orderMessage = JSON.parseObject(message, OrderMessage.class);        String userId = orderMessage.getUserId();                synchronized (userId.intern()) {            Long lastTime = lastProcessedTime.get(userId);            Long currentTime = orderMessage.getTimestamp();                        if (lastTime == null || currentTime >= lastTime) {                orderService.processOrder(orderMessage);                lastProcessedTime.put(userId, currentTime);            } else {                log.warn("消息时间戳错误，跳过: {}", orderMessage.getOrderId());            }        }    }}
复制代码


五、Spring Boot 实战应用

1. RabbitMQ 配置

@Configurationpublic class RabbitMQConfig {        @Bean    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();        factory.setConnectionFactory(connectionFactory());        factory.setConcurrentConsumers(1);  // 单线程消费，保证顺序        factory.setMaxConcurrentConsumers(1);        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);        return factory;    }        @Bean    public Queue orderQueue() {        return QueueBuilder.durable("order_queue")            .withArgument("x-message-ttl", 60000)            .build();    }}
复制代码

2. RocketMQ 配置

@Componentpublic class RocketMQConsumer {        @RocketMQMessageListener(        topic = "order_topic",        consumerGroup = "order_consumer_group",        consumeMode = ConsumeMode.ORDERLY,  // 顺序消费        messageModel = MessageModel.CLUSTERING    )    public class OrderConsumer implements RocketMQListener<String> {                @Autowired        private OrderService orderService;                @Override        public void onMessage(String message) {            OrderMessage orderMessage = JSON.parseObject(message, OrderMessage.class);            orderService.processOrder(orderMessage);        }    }}
复制代码

3. Kafka 配置

@Configurationpublic class KafkaConfig {        @Bean    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();        factory.setConsumerFactory(consumerFactory());        factory.setConcurrency(1);  // 单线程消费        factory.getContainerProperties().setPollTimeout(3000);        return factory;    }        @Bean    public ConsumerFactory<String, String> consumerFactory() {        Map<String, Object> props = new HashMap<>();        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order_group");        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");        return new DefaultKafkaConsumerFactory<>(props);    }}
复制代码


六、不同业务场景的解决方案

1. 电商系统

  • 订单创建：基于订单 ID 的幂等性，确保订单不重复创建。

  • 库存扣减：基于商品 ID 的幂等性，确保库存不重复扣减。

  • 支付处理：基于支付 ID 的幂等性，确保支付不重复处理。

  • 物流更新：基于物流单号的顺序性，确保物流状态正确更新。

2. 金融系统

  • 交易处理：基于交易 ID 的幂等性，确保交易不重复执行。

  • 余额更新：基于账户 ID 的顺序性，确保余额计算正确。

  • 风控检查：基于用户 ID 的顺序性，确保风控逻辑正确。

  • 对账处理：基于对账 ID 的幂等性，确保对账不重复执行。

3. 内容系统

  • 用户行为：基于用户 ID 的顺序性，确保行为轨迹正确。

  • 内容更新：基于内容 ID 的幂等性，确保内容不重复更新。

  • 推荐计算：基于用户 ID 的顺序性，确保推荐算法正确。

  • 统计计算：基于统计 ID 的幂等性，确保统计数据正确。


七、监控与告警

1. 监控指标

  • 重复消费次数：监控消息重复消费的次数。

  • 顺序错误次数：监控消息顺序错误的次数。

  • 处理延迟：监控消息处理的延迟时间。

  • 错误率：监控消息处理的错误率。

2. 告警策略

  • 重复消费告警：重复消费次数过多时告警。

  • 顺序错误告警：顺序错误次数过多时告警。

  • 延迟告警：处理延迟过长时告警。

  • 错误告警：错误率过高时告警。

3. 可视化面板

@RestController@RequestMapping("/admin/message")public class MessageAdminController {        @GetMapping("/stats")    public Map<String, Object> getMessageStats() {        Map<String, Object> stats = new HashMap<>();        stats.put("duplicateCount", getDuplicateCount());        stats.put("orderErrorCount", getOrderErrorCount());        stats.put("processDelay", getProcessDelay());        stats.put("errorRate", getErrorRate());        return stats;    }        @PostMapping("/reset")    public ResponseEntity<String> resetStats() {        // 重置统计信息        return ResponseEntity.ok("stats reset");    }}
复制代码


八、常见"坑"与优化建议

  1. 幂等性实现不当：幂等性逻辑有漏洞，导致重复消费。

  2. 顺序性保证不足：没有正确保证消息顺序，导致业务逻辑错误。

  3. 性能影响：幂等性检查影响性能，需要优化。

  4. 存储成本：幂等性标记占用存储空间，需要清理。

  5. 监控不到位：没有监控重复消费和顺序错误，无法及时发现问题。


九、最佳实践建议

  • 根据业务特点设计方案：不同业务有不同的幂等性和顺序性需求。

  • 监控和告警要到位：及时发现和处理异常情况。

  • 压测验证方案：通过压测验证方案的可行性。

  • 逐步优化和调整：根据实际运行情况，逐步优化方案。

  • 文档和规范要完善：建立消息处理规范和文档。


十、总结

消息重复消费和顺序性保障是分布式消息系统的核心问题，需要根据业务特点选择合适的解决方案。合理的实现和配置能够有效保证数据一致性和业务逻辑正确性。


关注服务端技术精选，获取更多后端实战干货！


你在消息重复消费和顺序性保障中遇到过哪些坑？欢迎在评论区分享你的故事！

划线
评论
复制
发布于: 刚刚阅读数: 2
消息队列重复消费顺序消费
用户头像

我爱娃哈哈😍

关注

个人博客： http://jiangyi.cool 2019-03-10 加入

公众号：服务端技术精选 欢迎大家关注！

评论

发布
暂无评论
Copyright © 2025, Geekbang Technology Ltd. All rights reserved. 极客邦控股（北京）有限公司 | 京 ICP 备 16027448 号 - 5京公网安备京公网安备 11010502039052号 | 产品资质
消息重复消费+顺序性，分布式消息的终极难题？一线解决方案全解析！_消息队列_我爱娃哈哈😍_InfoQ写作社区