Kafka 消息丢失的“防丢秘籍”：打造可靠的消息管道

Apache Kafka 的消息丢失可能导致业务问题，如订单漏处理。本文将以通俗易懂的方式，结合电商订单系统场景和 Go 语言代码示例，详细讲解消息丢失的原因和应对策略。内容适合 Kafka 初学者和进阶开发者。

为什么会发生消息丢失？

消息丢失发生在生产者、Broker或消费者端。

1. 生产者端丢失

异步发送未确认：
- 异步投递未确认，Broker 故障丢失。
- 场景：订单系统“下单”消息未确认，丢失。
确认级别不足：
- acks=0 或 acks=1，Leader 崩溃丢失。
- 场景：acks=0， “支付成功”未写入。
重试失败：
- retries 不足，网络中断丢失。
- 场景：网络抖动，retries=0，放弃投递。
缓冲区溢出：
- 缓冲区满，抛异常。
- 场景：订单高峰，消息被拒绝。

2. Broker 端丢失

副本不足：
- replication.factor=1，Broker 宕机丢失。
- 场景：单副本，磁盘故障全丢。
未同步副本：
- min.insync.replicas=1，Leader 宕机丢失。
- 场景：“支付成功”未同步，丢失。
日志清理：
- retention.ms 过期清理。
- 场景：消费者故障 2 天，订单消息清理。
磁盘故障：
- 磁盘损坏，无 RAID。
- 场景：Broker 坏块，数据丢失。

3. 消费者端丢失

自动提交提前：
- enable.auto.commit=true，崩溃前提交。
- 场景：“下单”提交后崩溃，未入库。
手动提交失败：
- 提交失败，重启跳过。
- 场景：“支付成功”提交失败，丢失。
消费滞后清理：
- Lag 过大，消息清理。
- 场景：滞后消息过期。
逻辑错误：
- 代码异常，跳过消息。
- 场景：解析失败，误跳过。

比喻：丢失像快递运输中丢包，需全程追踪。

应对消息丢失的策略

从生产者、Broker、消费者和系统级防止丢失。

1. 生产者端：确保投递可靠

同步/回调确认：
- 同步发送或回调确认。
- 场景：异步发送“下单”，回调记录失败。
- 配置：
  1
  
  acks=all
提高确认级别：
- acks=all, min.insync.replicas=2.
- 场景：“支付成功”写入 2 副本。
- 配置：
  1 2 3
  
  acks=all retries=3 retry.backoff.ms=100
增加重试：
- retries=3, retry.backoff.ms=100.
- 场景：网络抖动，重试投递。
扩大缓冲区：
- buffer.memory=67108864.
- 场景：高峰容纳消息。
- 配置：
  1 2
  
  buffer.memory=67108864 max.block.ms=60000
幂等性：
- enable.idempotence=true.
- 场景：防止重复投递丢失。
- 配置：
  1 2 3
  
  enable.idempotence=true acks=all retries=3

比喻：生产者像快递员加“投递确认”。

2. Broker 端：增强持久性

增加副本：

replication.factor=3.
场景：Broker 宕机，Follower 接管。

命令：

1

kafka-topics.sh --create --topic orders --bootstrap-server localhost:9092 --partitions 8 --replication-factor 3

同步副本：
- min.insync.replicas=2.
- 场景：“支付成功”同步 2 副本。
- Broker 配置：
  1 2
  
  default.replication.factor=3 min.insync.replicas=2

延长保留：

log.retention.ms=604800000.
场景：消费者故障 3 天不丢。

配置：

1

kafka-configs.sh --alter --entity-type topics --entity-name orders --bootstrap-server localhost:9092 --add-config retention.ms=604800000

日志压缩：

cleanup.policy=compact.
场景：订单状态保留最新。

配置：

1

kafka-configs.sh --alter --entity-type topics --entity-name orders_status --bootstrap-server localhost:9092 --add-config cleanup.policy=compact

硬件优化：
- RAID 10，log.flush.interval.messages=1.
- 场景：磁盘安全。

比喻：Broker 像仓库加“多备份”。

3. 消费者端：可靠处理

禁用自动提交：
- enable.auto.commit=false.
- 场景：入库后提交。
- 配置：
  1
  
  enable.auto.commit=false
同步提交：
- 同步提交 Offset。
- 场景：订单入库后提交。
处理异常：
- 捕获异常，存死信队列。
- 场景：解析失败，重试或存死信。
监控 Lag：
- 监控 kafka_consumer_lag.
- 场景：Lag 高，扩容消费者。
Exactly-Once：
- isolation.level=read_committed.
- 场景：订单事务化处理。
- 配置：
  1 2
  
  isolation.level=read_committed enable.auto.commit=false

比喻：消费者像收件人“核对签收”。

4. 系统级保障

监控告警：
- 监控 kafka_producer_record_error_rate, kafka_consumer_lag.
- 场景：Lag 激增，扩容。
死信队列：
- 存失败消息到 orders_dead_letter.
- 场景：解析失败存死信。
冗余部署：
- 3 Broker，KRaft 模式。
- 场景：元数据可靠。
压测演练：
- 模拟宕机，验证不丢。
- 场景：关闭 Broker，确认投递。

比喻：系统级像“监控中心”。

策略对比与选择

策略	优点	缺点	适用场景
生产者同步/回调确认	确保投递	同步增加延迟	高可靠性场景
生产者 `acks=all`	多副本写入	增加延迟	订单、支付
生产者幂等性	防重复	配置复杂	高并发，网络不稳
Broker 多副本	容错性强	增加存储	所有生产环境
Broker 延长保留	防清理	增加磁盘	消费者滞后
消费者手动提交	确保处理	增加复杂性	通用场景
消费者 Exactly-Once	强一致性	性能开销	金融、订单
系统监控与死信队列	可追溯	需开发	复杂业务

场景选择：

订单系统：acks=all + 幂等性 + 多副本 + 手动提交 + 死信队列。
金融：Exactly-Once。
日志：acks=1，延长保留。

比喻：策略像选“防丢套餐”。

优化与监控

优化策略：

生产者：
- batch.size=65536, linger.ms=10.
- compression.type=snappy.
Broker：
- num.io.threads=8.
- log.flush.interval.messages=1（谨慎）。
消费者：
- max.poll.records=100.
- session.timeout.ms=10000.
监控：
- kafka_producer_record_error_rate.
- kafka_consumer_lag.
- 工具：Prometheus + Grafana.

死信队列：

创建 orders_dead_letter：

1

kafka-topics.sh --create --topic orders_dead_letter --bootstrap-server localhost:9092 --partitions 8 --replication-factor 3

演练：
- 模拟宕机，验证不丢。

比喻：优化像加“智能追踪”。

代码示例：可靠生产与消费

以下 Go 程序实现可靠订单处理。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146


package main

import (
	"encoding/json"
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"log"
	"time"
)

// Order 表示订单消息
type Order struct {
	OrderID   string `json:"order_id"`
	Amount    int    `json:"amount"`
	Timestamp int64  `json:"timestamp"`
}

// 生产者：可靠投递订单
func produceOrders() {
	config := &kafka.ConfigMap{
		"bootstrap.servers":        "localhost:9092",
		"acks":                     "all",
		"enable.idempotence":       true,
		"retries":                  3,
		"retry.backoff.ms":         100,
		"buffer.memory":            67108864, // 64MB
		"batch.size":               65536,    // 64KB
		"linger.ms":                10,
		"compression.type":         "snappy",
		"max.in.flight.requests.per.connection": 5,
	}

	producer, err := kafka.NewProducer(config)
	if err != nil {
		log.Fatalf("Failed to create producer: %v", err)
	}
	defer producer.Close()

	topic := "orders"
	deliveryChan := make(chan kafka.Event, 100)

	// 发送 1000 条订单
	for i := 0; i < 1000; i++ {
		order := Order{
			OrderID:   fmt.Sprintf("order%d", i),
			Amount:    100 + i,
			Timestamp: time.Now().UnixMilli(),
		}
		value, _ := json.Marshal(order)

		err = producer.Produce(&kafka.Message{
			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
			Key:            []byte(order.OrderID),
			Value:          value,
		}, deliveryChan)
		if err != nil {
			log.Printf("Failed to produce order %s: %v", order.OrderID, err)
			continue
		}

		// 确认投递
		e := <-deliveryChan
		msg := e.(*kafka.Message)
		if msg.TopicPartition.Error != nil {
			log.Printf("Delivery failed for %s: %v", order.OrderID, msg.TopicPartition.Error)
		} else {
			log.Printf("Delivered order %s to %s", order.OrderID, msg.TopicPartition)
		}
	}

	producer.Flush(5000)
}

// 消费者：可靠处理订单
func consumeOrders() {
	config := &kafka.ConfigMap{
		"bootstrap.servers":   "localhost:9092",
		"group.id":            "orders-group",
		"auto.offset.reset":   "earliest",
		"enable.auto.commit":  false,
		"session.timeout.ms":  10000,
		"max.poll.records":    100,
		"isolation.level":     "read_committed",
	}

	consumer, err := kafka.NewConsumer(config)
	if err != nil {
		log.Fatalf("Failed to create consumer: %v", err)
	}
	defer consumer.Close()

	err = consumer.SubscribeTopics([]string{"orders"}, nil)
	if err != nil {
		log.Fatalf("Failed to subscribe: %v", err)
	}

	// 死信队列生产者
	dlqProducer, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost:9092"})
	if err != nil {
		log.Fatalf("Failed to create DLQ producer: %v", err)
	}
	defer dlqProducer.Close()
	dlqTopic := "orders_dead_letter"

	for {
		msg, err := consumer.ReadMessage(time.Second * 10)
		if err != nil {
			log.Printf("Consumer error: %v", err)
			continue
		}

		var order Order
		if err := json.Unmarshal(msg.Value, &order); err != nil {
			log.Printf("Failed to unmarshal order: %v", err)
			// 存入死信队列
			dlqProducer.Produce(&kafka.Message{
				TopicPartition: kafka.TopicPartition{Topic: &dlqTopic, Partition: kafka.PartitionAny},
				Key:            msg.Key,
				Value:          msg.Value,
			}, nil)
			consumer.CommitMessage(msg)
			continue
		}

		// 模拟处理（入库）
		log.Printf("Processing order: %s, amount=%d", order.OrderID, order.Amount)
		// 假设处理成功

		// 同步提交 Offset
		_, err = consumer.CommitMessage(msg)
		if err != nil {
			log.Printf("Failed to commit offset for %s: %v", order.OrderID, err)
			// 可记录到死信队列或重试
			continue
		}
	}
}

func main() {
	// 启动生产者
	go produceOrders()
	time.Sleep(time.Second * 5)

	// 启动消费者
	consumeOrders()
}

代码说明

生产者：
- acks=all, enable.idempotence=true.
- 批量优化：batch.size, linger.ms.
- 确认投递，记录失败。
消费者：
- enable.auto.commit=false.
- 同步提交，存死信队列。
死信队列：
- 失败消息存 orders_dead_letter.
日志：
- 记录投递和处理。

运行准备

安装 Kafka：

运行 Kafka（端口 9092）、ZooKeeper（端口 2181）。

创建主题：

1
2


kafka-topics.sh --create --topic orders --bootstrap-server localhost:9092 --partitions 8 --replication-factor 3 --config retention.ms=604800000
kafka-topics.sh --create --topic orders_dead_letter --bootstrap-server localhost:9092 --partitions 8 --replication-factor 3

配置 Broker：

1
2
3


default.replication.factor=3
min.insync.replicas=2
log.flush.interval.messages=1

安装依赖：

1

go get github.com/confluentinc/confluent-kafka-go/kafka

运行：

go run kafka_reliable_order.go

输出：

Delivered order order1 to orders[3]
Processing order: order1, amount=101

验证：

1

kafka-console-consumer.sh --topic orders --bootstrap-server localhost:9092 --from-beginning

扩展建议

集成 MySQL。
监控 Lag 和错误率。
Exactly-Once 事务。
压测高并发。

注意事项与最佳实践

生产者：
- acks=all, enable.idempotence=true.
- buffer.memory=67108864.
Broker：
- replication.factor=3, min.insync.replicas=2.
- log.retention.ms=604800000.
消费者：
- enable.auto.commit=false.
- 死信队列。
监控：
- kafka_consumer_lag.
- 告警 Lag > 100 万。
KRaft：
- 测试 KRaft。
压测：
- 模拟宕机。

比喻：防丢失像“全程保险”。

总结

Kafka 消息丢失由生产者未确认、Broker 副本不足、消费者提交错误等引发，可通过生产者 acks=all、Broker 多副本、消费者手动提交和系统监控解决。本文结合订单系统场景和 Go 代码示例，讲解了原理和实践。希望这篇文章帮助你掌握消息丢失的“防丢秘籍”，并在生产环境中应用！

如需更多问题或补充，欢迎留言讨论。

linying

文章目录