明明發送了重要消息，但消費者就是收不到？或者消息處理了一半，突然就消失了？這些問題很可能就是Kafka消息丟失造成的！今天就來聊聊Kafka消息丟失的3種典型場景，以及如何在生產環境中完美避免這些坑！
一、Kafka消息丟失的根源分析
在深入討論具體場景之前，我們先來了解一下Kafka的消息流轉過程和可能導致消息丟失的關鍵環節。
1.1 Kafka消息流轉過程
// Kafka消息流轉過程簡述
public class KafkaMessageFlow {
public void flow() {
System.out.println("=== Kafka消息流轉過程 ===");
System.out.println("1. Producer發送消息到Broker");
System.out.println("2. Broker將消息寫入磁盤並確認");
System.out.println("3. Consumer從Broker拉取消息");
System.out.println("4. Consumer處理消息並提交偏移量");
System.out.println("5. 消息被成功消費");
}
}
1.2 消息丟失的關鍵環節
// 消息丟失的關鍵環節
public class MessageLossPoints {
public void lossPoints() {
System.out.println("=== 消息丟失的關鍵環節 ===");
System.out.println("1. Producer發送階段：網絡問題、Broker不可用");
System.out.println("2. Broker存儲階段：磁盤故障、副本同步失敗");
System.out.println("3. Consumer消費階段：處理失敗、偏移量提交異常");
}
}
二、場景一：Producer發送消息丟失
這是最常見的消息丟失場景，特別是在網絡不穩定或Broker負載過高時容易出現。
2.1 問題表現
// Producer發送消息丟失的表現
public class ProducerLossSymptoms {
public void symptoms() {
System.out.println("=== Producer發送消息丟失的表現 ===");
System.out.println("1. send()方法返回成功，但消費者收不到消息");
System.out.println("2. 消息在Broker中找不到");
System.out.println("3. 監控顯示發送成功率100%，但實際消息缺失");
}
}
2.2 錯誤的配置示例
// 錯誤的Producer配置示例
@Configuration
public class WrongKafkaProducerConfig {
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// 錯誤配置1：acks設置為0，不等待任何確認
props.put(ProducerConfig.ACKS_CONFIG, "0");
// 錯誤配置2：重試次數為0，不重試
props.put(ProducerConfig.RETRIES_CONFIG, 0);
// 錯誤配置3：沒有設置消息確認超時
// props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
2.3 正確的解決方案
// 正確的Producer配置
@Configuration
public class CorrectKafkaProducerConfig {
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// 正確配置1：acks=all，等待所有副本確認
props.put(ProducerConfig.ACKS_CONFIG, "all");
// 正確配置2：設置重試次數
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
// 正確配置3：開啓冪等性
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// 正確配置4：設置消息確認超時
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 60000);
// 正確配置5：設置批次大小和linger時間
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
// 設置默認Topic
kafkaTemplate.setDefaultTopic("default-topic");
return kafkaTemplate;
}
}
2.4 安全的消息發送方式
@Service
@Slf4j
public class SafeMessageSender {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
/**
* 同步發送消息（確保消息送達）
*/
public void sendSync(String topic, String key, String message) {
try {
SendResult<String, String> result = kafkaTemplate.send(topic, key, message).get(10, TimeUnit.SECONDS);
log.info("消息發送成功: topic={}, partition={}, offset={}",
result.getRecordMetadata().topic(),
result.getRecordMetadata().partition(),
result.getRecordMetadata().offset());
} catch (Exception e) {
log.error("消息發送失敗: topic={}, key={}, message={}", topic, key, message, e);
throw new RuntimeException("消息發送失敗", e);
}
}
/**
* 異步發送消息（帶回調確認）
*/
public void sendAsync(String topic, String key, String message) {
ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, message);
future.addCallback(
new ListenableFutureCallback<SendResult<String, String>>() {
@Override
public void onSuccess(SendResult<String, String> result) {
log.info("消息發送成功: topic={}, partition={}, offset={}",
result.getRecordMetadata().topic(),
result.getRecordMetadata().partition(),
result.getRecordMetadata().offset());
}
@Override
public void onFailure(Throwable ex) {
log.error("消息發送失敗: topic={}, key={}, message={}", topic, key, message, ex);
// 可以在這裏實現重試邏輯
handleSendFailure(topic, key, message, ex);
}
}
);
}
/**
* 處理髮送失敗的情況
*/
private void handleSendFailure(String topic, String key, String message, Throwable ex) {
// 記錄到數據庫或發送到死信隊列
log.warn("準備重試發送消息: topic={}, key={}", topic, key);
// 實現重試邏輯（可以使用Spring Retry或其他重試框架）
retrySendMessage(topic, key, message, ex);
}
private void retrySendMessage(String topic, String key, String message, Throwable ex) {
// 重試邏輯實現
// 可以記錄到數據庫，由定時任務重試
// 或者直接再次發送
}
}
三、場景二：Broker存儲消息丟失
當Broker節點發生故障或磁盤損壞時，可能會導致已接收但未完全同步的消息丟失。
3.1 問題表現
// Broker存儲消息丟失的表現
public class BrokerLossSymptoms {
public void symptoms() {
System.out.println("=== Broker存儲消息丟失的表現 ===");
System.out.println("1. Broker節點宕機後，部分消息無法找回");
System.out.println("2. 副本同步失敗，Leader切換後消息缺失");
System.out.println("3. 磁盤故障導致部分分區數據丟失");
}
}
3.2 錯誤的Broker配置
# 錯誤的Broker配置示例 server.properties
# 錯誤配置1：副本因子過小
num.replica.fetchers=1
default.replication.factor=1 # 單副本，無冗餘
# 錯誤配置2：刷盤策略過於寬鬆
log.flush.interval.messages=9223372036854775807 # 幾乎不刷盤
log.flush.interval.ms=9223372036854775807 # 幾乎不刷盤
# 錯誤配置3：未開啓自動Leader選舉
auto.leader.rebalance.enable=false
3.3 正確的Broker配置
# 正確的Broker配置 server.properties
# 正確配置1：合理的副本因子
default.replication.factor=3 # 至少3副本保證高可用
min.insync.replicas=2 # 至少2個副本確認才算成功
# 正確配置2：嚴格的刷盤策略
log.flush.interval.messages=10000 # 每1萬條消息刷盤一次
log.flush.interval.ms=1000 # 每秒刷盤一次
# 正確配置3：開啓自動Leader選舉
auto.leader.rebalance.enable=true
# 正確配置4：其他重要配置
unclean.leader.election.enable=false # 禁止不乾淨的Leader選舉
retention.ms=604800000 # 消息保留7天
3.4 監控和告警配置
// Broker監控配置
@Component
@Slf4j
public class KafkaBrokerMonitor {
@Autowired
private MeterRegistry meterRegistry;
/**
* 監控副本同步狀態
*/
public void monitorReplicaSync() {
// 監控Under Replicated Partitions
Gauge.builder("kafka.broker.under.replicated.partitions")
.description("未完全同步的分區數量")
.register(meterRegistry, this, KafkaBrokerMonitor::getUnderReplicatedPartitions);
// 監控Offline Partitions
Gauge.builder("kafka.broker.offline.partitions")
.description("離線分區數量")
.register(meterRegistry, this, KafkaBrokerMonitor::getOfflinePartitions);
}
private double getUnderReplicatedPartitions() {
// 獲取未完全同步的分區數量
// 可以通過JMX或Kafka AdminClient獲取
return 0;
}
private double getOfflinePartitions() {
// 獲取離線分區數量
return 0;
}
/**
* 告警檢查
*/
@Scheduled(fixedRate = 60000) // 每分鐘檢查一次
public void checkBrokerHealth() {
int underReplicated = (int) getUnderReplicatedPartitions();
int offlinePartitions = (int) getOfflinePartitions();
if (underReplicated > 0) {
log.warn("發現{}個未完全同步的分區", underReplicated);
// 發送告警
sendAlert("Kafka副本同步異常", "發現" + underReplicated + "個未完全同步的分區");
}
if (offlinePartitions > 0) {
log.error("發現{}個離線分區", offlinePartitions);
// 發送嚴重告警
sendCriticalAlert("Kafka分區離線", "發現" + offlinePartitions + "個離線分區");
}
}
private void sendAlert(String title, String message) {
// 告警發送實現
}
private void sendCriticalAlert(String title, String message) {
// 嚴重告警發送實現
}
}
四、場景三：Consumer消費消息丟失
這是最容易被忽視的場景，特別是在消費者處理邏輯複雜或異常處理不當的情況下。
4.1 問題表現
// Consumer消費消息丟失的表現
public class ConsumerLossSymptoms {
public void symptoms() {
System.out.println("=== Consumer消費消息丟失的表現 ===");
System.out.println("1. 消息被消費但業務邏輯未執行完成");
System.out.println("2. 異常處理不當導致消息被跳過");
System.out.println("3. 偏移量提前提交導致消息丟失");
System.out.println("4. 消費者重啓後重復消費或丟失消息");
}
}
4.2 錯誤的Consumer實現
// 錯誤的Consumer實現示例
@Component
@Slf4j
public class WrongMessageConsumer {
// 錯誤實現1：自動提交偏移量
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void consumeWrong(ConsumerRecord<String, String> record) {
log.info("收到消息: topic={}, partition={}, offset={}, key={}, value={}",
record.topic(), record.partition(), record.offset(), record.key(), record.value());
try {
// 處理業務邏輯
processBusinessLogic(record.value());
// 錯誤實現2：處理完就認為成功，不管是否真的成功
log.info("消息處理完成");
} catch (Exception e) {
// 錯誤實現3：異常處理不當，消息丟失
log.error("處理消息失敗", e);
// 沒有重試機制，消息直接丟失
}
// 錯誤實現4：這裏已經是自動提交偏移量了（如果配置為自動提交）
}
private void processBusinessLogic(String message) throws Exception {
// 模擬複雜的業務處理邏輯
// 可能拋出異常
if (Math.random() < 0.1) {
throw new RuntimeException("模擬業務處理異常");
}
// 模擬長時間處理
Thread.sleep(1000);
}
}
4.3 正確的Consumer實現
// 正確的Consumer實現
@Component
@Slf4j
public class CorrectMessageConsumer {
@Autowired
private BusinessService businessService;
@Autowired
private DeadLetterQueueService deadLetterQueueService;
// 正確實現1：手動提交偏移量
@KafkaListener(
topics = "order-topic",
groupId = "order-group",
containerFactory = "kafkaListenerContainerFactory"
)
public void consumeCorrect(ConsumerRecord<String, String> record, Acknowledgment ack) {
String topic = record.topic();
int partition = record.partition();
long offset = record.offset();
String key = record.key();
String value = record.value();
log.info("收到消息: topic={}, partition={}, offset={}, key={}", topic, partition, offset, key);
try {
// 正確實現2：完整的業務處理流程
boolean success = processBusinessLogic(value);
if (success) {
// 正確實現3：處理成功後再提交偏移量
ack.acknowledge();
log.info("消息處理成功並確認: offset={}", offset);
} else {
// 處理失敗，發送到死信隊列
handleProcessingFailure(record, ack);
}
} catch (Exception e) {
// 正確實現4：完善的異常處理機制
log.error("處理消息失敗: topic={}, partition={}, offset={}", topic, partition, offset, e);
handleProcessingException(record, ack, e);
}
}
/**
* 處理業務邏輯
*/
private boolean processBusinessLogic(String message) {
try {
// 調用業務服務處理消息
businessService.handleMessage(message);
return true;
} catch (Exception e) {
log.error("業務處理異常", e);
return false;
}
}
/**
* 處理業務處理失敗的情況
*/
private void handleProcessingFailure(ConsumerRecord<String, String> record, Acknowledgment ack) {
log.warn("業務處理失敗，發送到死信隊列: offset={}", record.offset());
// 發送到死信隊列
deadLetterQueueService.sendToDLQ(record);
// 確認原消息（因為我們已經將其發送到DLQ）
ack.acknowledge();
}
/**
* 處理異常情況
*/
private void handleProcessingException(ConsumerRecord<String, String> record, Acknowledgment ack, Exception e) {
// 記錄異常信息
log.error("消息處理異常: offset={}, exception={}", record.offset(), e.getMessage());
// 根據異常類型決定處理策略
if (isRetriableException(e)) {
// 可重試異常，不確認偏移量，讓Kafka重新投遞
log.info("可重試異常，消息將被重新投遞: offset={}", record.offset());
// 不調用ack.acknowledge()，消息會被重新消費
} else {
// 不可重試異常，發送到死信隊列
handleProcessingFailure(record, ack);
}
}
/**
* 判斷是否為可重試異常
*/
private boolean isRetriableException(Exception e) {
// 可以根據具體業務需求定義哪些異常是可重試的
return e instanceof TimeoutException ||
e instanceof RetriableException ||
e instanceof NetworkException;
}
}
4.4 Consumer配置優化
// 正確的Consumer配置
@Configuration
public class CorrectKafkaConsumerConfig {
@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);
// 正確配置1：手動提交偏移量
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
// 正確配置2：設置合理的會話超時
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000);
// 正確配置3：設置最大拉取大小
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1048576); // 1MB
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
// 正確配置4：設置重試間隔
props.put(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG, 1000);
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
// 正確配置5：設置手動確認模式
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
// 正確配置6：設置併發消費者數量
factory.setConcurrency(3);
// 正確配置7：設置錯誤處理器
factory.setErrorHandler(new SeekToCurrentErrorHandler(new FixedBackOff(1000L, 3L)));
return factory;
}
}
五、綜合保障措施
5.1 消息追蹤和監控
// 消息追蹤實現
@Component
@Slf4j
public class MessageTraceService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
/**
* 記錄消息軌跡
*/
public void traceMessage(String messageId, String status, String info) {
String traceKey = "message_trace:" + messageId;
String traceInfo = System.currentTimeMillis() + "|" + status + "|" + info;
redisTemplate.opsForList().leftPush(traceKey, traceInfo);
redisTemplate.expire(traceKey, Duration.ofHours(24));
}
/**
* 查詢消息軌跡
*/
public List<String> getMessageTrace(String messageId) {
String traceKey = "message_trace:" + messageId;
return redisTemplate.opsForList().range(traceKey, 0, -1);
}
}
5.2 死信隊列處理
// 死信隊列處理
@Component
@Slf4j
public class DeadLetterQueueService {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Autowired
private FailedMessageRepository failedMessageRepository;
/**
* 發送到死信隊列
*/
public void sendToDLQ(ConsumerRecord<String, String> record) {
try {
// 構造死信消息
FailedMessage failedMessage = new FailedMessage();
failedMessage.setTopic(record.topic());
failedMessage.setPartition(record.partition());
failedMessage.setOffset(record.offset());
failedMessage.setKey(record.key());
failedMessage.setValue(record.value());
failedMessage.setTimestamp(System.currentTimeMillis());
failedMessage.setException("處理失敗");
// 保存到數據庫
failedMessageRepository.save(failedMessage);
// 發送到死信Topic
String dlqTopic = record.topic() + ".dlq";
kafkaTemplate.send(dlqTopic, record.key(), record.value());
log.info("消息已發送到死信隊列: topic={}, offset={}", record.topic(), record.offset());
} catch (Exception e) {
log.error("發送死信消息失敗", e);
}
}
/**
* 死信消息處理
*/
@KafkaListener(topics = "order-topic.dlq", groupId = "dlq-group")
public void handleDLQMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
log.info("收到死信消息: topic={}, offset={}", record.topic(), record.offset());
try {
// 嘗試重新處理
reprocessMessage(record);
// 確認消息
ack.acknowledge();
} catch (Exception e) {
log.error("死信消息處理失敗", e);
// 可以記錄到人工處理隊列
moveToManualQueue(record);
ack.acknowledge();
}
}
private void reprocessMessage(ConsumerRecord<String, String> record) {
// 重新處理邏輯
}
private void moveToManualQueue(ConsumerRecord<String, String> record) {
// 移動到人工處理隊列
}
}
5.3 完整的監控告警體系
// 完整的監控告警體系
@Component
@Slf4j
public class KafkaMonitoringService {
@Autowired
private MeterRegistry meterRegistry;
private final Counter messageProducedCounter;
private final Counter messageConsumedCounter;
private final Counter messageLostCounter;
private final Timer messageProcessTimer;
public KafkaMonitoringService(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
this.messageProducedCounter = Counter.builder("kafka.messages.produced")
.description("已生產的消息數量")
.register(meterRegistry);
this.messageConsumedCounter = Counter.builder("kafka.messages.consumed")
.description("已消費的消息數量")
.register(meterRegistry);
this.messageLostCounter = Counter.builder("kafka.messages.lost")
.description("丟失的消息數量")
.register(meterRegistry);
this.messageProcessTimer = Timer.builder("kafka.message.process.time")
.description("消息處理耗時")
.register(meterRegistry);
}
/**
* 記錄消息生產
*/
public void recordMessageProduced() {
messageProducedCounter.increment();
}
/**
* 記錄消息消費
*/
public void recordMessageConsumed(long processTimeMs) {
messageConsumedCounter.increment();
messageProcessTimer.record(processTimeMs, TimeUnit.MILLISECONDS);
}
/**
* 記錄消息丟失
*/
public void recordMessageLost() {
messageLostCounter.increment();
// 發送告警
sendLossAlert();
}
private void sendLossAlert() {
log.error("檢測到Kafka消息丟失！");
// 實現告警發送邏輯
}
}
六、生產環境最佳實踐
6.1 部署架構建議
// 生產環境部署架構建議
public class ProductionDeploymentGuide {
public void guide() {
System.out.println("=== Kafka生產環境部署建議 ===");
System.out.println("1. Broker節點：至少3個節點，分佈在不同機架");
System.out.println("2. Zookeeper：奇數個節點（3或5），獨立部署");
System.out.println("3. 磁盤：SSD存儲，單獨掛載數據目錄");
System.out.println("4. 網絡：千兆網絡，低延遲");
System.out.println("5. 監控：Prometheus + Grafana + AlertManager");
}
}
6.2 運維操作手冊
// 運維操作手冊
public class OperationsManual {
public void manual() {
System.out.println("=== Kafka運維操作手冊 ===");
System.out.println("日常檢查：");
System.out.println("- 監控Broker狀態和資源使用率");
System.out.println("- 檢查副本同步狀態");
System.out.println("- 監控消費者組滯後情況");
System.out.println("- 檢查磁盤空間使用情況");
System.out.println("\n應急處理：");
System.out.println("- Broker宕機：檢查日誌，重啓服務");
System.out.println("- 分區離線：檢查副本狀態，重新分配分區");
System.out.println("- 消費者滯後：增加消費者實例");
System.out.println("- 磁盤滿：清理舊數據，擴容磁盤");
System.out.println("\n預防措施：");
System.out.println("- 定期備份重要數據");
System.out.println("- 配置合理的監控告警");
System.out.println("- 制定災難恢復預案");
System.out.println("- 定期進行壓力測試");
}
}
結語
Kafka消息丟失問題是生產環境中必須高度重視的問題。通過本文介紹的三種典型場景和相應的解決方案，相信你能有效避免消息丟失的風險。
關鍵要點總結：
- Producer端：使用acks=all、開啓冪等性、同步發送或帶回調的異步發送
- Broker端：配置合理的副本因子、刷盤策略、監控副本同步狀態
- Consumer端：手動提交偏移量、完善的異常處理機制、死信隊列處理
記住，消息系統的可靠性不是自然而然的，需要我們在每個環節都做好充分的保障措施。在分佈式系統中，任何組件都可能出現故障，關鍵是要有完善的容錯和恢復機制。
如果你覺得這篇文章對你有幫助，歡迎分享給更多的朋友。在Kafka消息系統的設計和運維路上，我們一起成長！
關注「服務端技術精選」，獲取更多幹貨技術文章！
