????????在 Java 分布式系統開發中，消息隊列的應用已十分普遍。但隨著業務規模擴大，消息的重復消費、意外消失、順序錯亂等問題逐漸成為系統穩定性的隱患。本文將從 Java 開發者的視角，深入分析這三大問題的產生原因、業務后果，并結合具體代碼示例給出可落地的解決方案。?
消息重復消費是 Java 開發中最易遇到的問題。例如，使用 Spring Kafka 消費支付消息時，若處理邏輯未做防護，可能導致用戶賬戶被重復扣款，這在金融場景中是致命的。?
產生原因通常有以下幾點
生產者重試配置不當：例如：生產者使用RetryTemplate時，若未設置合理的重試條件，網絡波動時會重復發送消息。?
消費者 ACK 機制誤用：例如：Spring Kafka 默認ack-mode=BATCH，若消費者處理消息后未及時提交 offset，重啟后會重復消費批次內消息。?
分布式事務補償：例如：在 Seata 等分布式事務框架中，回滾后觸發的消息重發可能導致重復。?
在不同場景下，帶來的后果也十分嚴重，如在金融領域會導致出現重復扣款、重復轉賬的問題，從而引發資損和用戶投訴。?在庫存管理系統中會導致重復扣減庫存，從而出現超賣或負庫存。為了解決這個? ? ?
? ? ? ? 通常采用多種方法解決，如：
? ? ? ? （1）冪等性設計
? ? ? ? （2全局 ID+Redis 去重
@Service
public class OrderConsumer {@Autowiredprivate StringRedisTemplate redisTemplate;@Autowiredprivate OrderService orderService;@KafkaListener(topics = "order_pay")public void handlePayMessage(ConsumerRecord<String, String> record) {String messageId = record.headers().lastHeader("messageId").value().toString();// 利用Redis的SETNX判斷是否已處理Boolean isFirstHandle = redisTemplate.opsForValue().setIfAbsent("msg:processed:" + messageId, "1", 24, TimeUnit.HOURS);if (Boolean.TRUE.equals(isFirstHandle)) {// 首次處理：執行業務邏輯orderService.processPayment(record.value());} else {// 重復消息：直接返回log.info("重復消息，messageId:{}", messageId);}}
}? ? ? ? （3）數據庫唯一約束
@Transactional
public void processPayment(String orderJson) {OrderDTO order = JSON.parseObject(orderJson, OrderDTO.class);// 插入時若messageId重復，會拋出DuplicateKeyExceptionorderMapper.insert(new OrderPO().setOrderId(order.getOrderId()).setMessageId(order.getMessageId()).setStatus("PAID"));
}
消息消失指消息未被消費卻永久丟失，例如用戶下單消息消失會導致訂單 “幽靈下單”，用戶已付款但系統無記錄。?
產生原因通常有以下幾點
生產者未開啟確認機制：如Kafka 生產者未設置acks=all，消息未寫入分區副本即返回成功。?
Spring 容器關閉導致丟失：消費者在@PreDestroy階段被強制關閉，未處理的消息被丟棄。?
中間件配置疏漏：如RabbitMQ 隊列未設置durable=true，重啟后隊列消失；Kafka 未設置log.retention.hours，消息提前過期。?
其也會導致許多嚴重的后果，如：
交易鏈路斷裂：支付成功但訂單狀態未更新，用戶投訴。?
數據同步失敗：跨系統數據未同步，導致庫存、會員信息不一致。?
任務調度失效：定時任務觸發消息丟失，導致任務漏執行。?
通常采用如下的解決方案：全鏈路可靠性保障?
1. 生產者確認機制（
@Configuration?
public class KafkaProducerConfig {?@Bean?public ProducerFactory<String, String> producerFactory() {?Map<String, Object> config = new HashMap<>();?config.put(ProducerConfig.ACKS_CONFIG, "all"); // 等待所有副本確認?config.put(ProducerConfig.RETRIES_CONFIG, 3); // 重試3次?config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 開啟冪等性生產?return new DefaultKafkaProducerFactory<>(config);?}?
?@Bean?public KafkaTemplate<String, String> kafkaTemplate() {?KafkaTemplate<String, String> template = new KafkaTemplate<>(producerFactory());?// 發送結果回調?template.setProducerListener(new ProducerListener<>() {?@Override?public void onError(ProducerRecord<String, String> record, Exception exception) {?log.error("消息發送失敗，topic:{}, msg:{}", record.topic(), record.value(), exception);?// 失敗后可存入本地消息表，定時重試?}?});?sdareturn template;?}?
}?
?
2. 消費者手動確認
?
@Configuration?
public class KafkaConsumerConfig {?@Bean?public ConsumerFactory<String, String> consumerFactory() {?Map<String, Object> config = new HashMap<>();?config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 關閉自動提交?return new DefaultKafkaConsumerFactory<>(config);?}?
?@Bean?public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {?ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();?factory.setConsumerFactory(consumerFactory());?factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); // 手動確認?return factory;?}?
}?
?
// 消費者代碼?
@KafkaListener(topics = "order_create")?
public void handleCreateMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {?try {?orderService.createOrder(record.value());?ack.acknowledge(); // 處理成功：手動提交offset?} catch (Exception e) {?log.error("處理失敗，暫不確認", e);?// 可將消息轉發到死信隊列?}?
}?
消息順序性指消息處理順序與發送順序一致。例如，訂單的 “創建→支付→發貨” 消息若順序錯亂，會導致 “未支付就發貨” 的邏輯錯誤。?產生原因通常有以下幾點：
多線程消費：Spring Kafka 的concurrency>1時，多個線程并行處理同一分區消息。?
分區路由錯誤：Kafka 生產者未指定partitioner.class，導致同一訂單的消息被分配到不同分區。?
重試機制打亂順序：失敗消息進入重試隊列后，后續消息先被處理。?
其可能會導致許多后果，如：
狀態機異常：訂單從 “待支付” 直接跳至 “已發貨”，狀態流轉斷裂。?
數據計算錯誤：賬戶先扣款后充值，導致余額計算錯誤。?
日志時序混亂：分布式追蹤日志順序錯亂，難以排查問題。?
通常開發人員采用以下方案解決
1. 單分區 + 單線程消費
2. 按業務 ID 路由分區?
通過訂單 ID 哈希到固定分區，確保同一訂單的消息在同一分區：?
?
public class OrderPartitioner implements Partitioner {?@Override?public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {?// 訂單ID作為key，哈希后取模分區數?String orderId = (String) key;?List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);?return Math.abs(orderId.hashCode()) % partitions.size();?}?
}?
)
