一、Kafka消息传输的默认行为
Kafka作为分布式消息队列,默认的消息传输机制其实存在几个典型问题。我们先来看一个生产者示例(技术栈:Java):
// 创建生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 创建生产者实例
Producer<String, String> producer = new KafkaProducer<>(props);
// 发送消息(默认异步发送)
producer.send(new ProducerRecord<>("test-topic", "message-key", "hello world"));
// 不保证消息到达就关闭生产者
producer.close();
这里暴露了两个问题:
- 发送模式默认是异步的，不等待broker确认
- 生产者关闭时没有调用flush()，可能丢失在缓冲区中的消息
二、消息丢失的三大场景
2.1 生产者端丢失
当网络抖动时，默认配置下生产者不会自动重试。改进方案：
// 增加可靠性配置
props.put("acks", "all"); // 需要所有ISR副本确认
props.put("retries", 3); // 重试次数
props.put("retry.backoff.ms", 300); // 重试间隔
// 同步发送确保消息到达
Future<RecordMetadata> future = producer.send(
new ProducerRecord<>("test-topic", "message-key", "hello world"));
future.get(); // 阻塞等待发送结果
2.2 Broker端丢失
当副本不同步且leader崩溃时可能丢失数据。关键配置：
# broker配置
unclean.leader.election.enable=false # 禁止不同步副本成为leader
min.insync.replicas=2 # 最小同步副本数
2.3 消费者端丢失
默认自动提交offset可能导致重复消费：
// 消费者配置示例
props.put("enable.auto.commit", "false"); // 关闭自动提交
// 手动提交offset
while(true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
// 处理消息
processMessage(record);
}
consumer.commitSync(); // 同步提交
}
三、消息顺序性保障
Kafka默认只能保证分区内顺序。典型乱序场景：
// 错误示例：异步发送不保证顺序
for(int i=0; i<100; i++){
producer.send(new ProducerRecord<>("ordered-topic", "key", "msg-"+i));
}
// 正确做法：同步发送或使用单线程
for(int i=0; i<100; i++){
producer.send(new ProducerRecord<>("ordered-topic", "key", "msg-"+i)).get();
}
对于需要严格顺序的场景，还可以：
- 设置max.in.flight.requests.per.connection=1
- 使用相同消息key确保路由到同一分区
四、消息积压解决方案
当消费者处理速度跟不上时，可以：
4.1 水平扩展
// 增加消费者实例数(需小于等于分区数)
props.put("group.id", "consumer-group");
KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(props);
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(props);
// 订阅相同topic
consumer1.subscribe(Collections.singletonList("busy-topic"));
consumer2.subscribe(Collections.singletonList("busy-topic"));
4.2 调整消费参数
# 提高吞吐量配置
fetch.max.bytes=52428800 # 单次拉取最大字节数
max.poll.records=500 # 单次拉取最大消息数
4.3 死信队列处理
// 创建死信生产者
Producer<String, String> dlqProducer = new KafkaProducer<>(props);
try {
processMessage(record);
} catch (Exception e) {
// 发送到死信队列
dlqProducer.send(new ProducerRecord<>("dlq-topic", record.key(), record.value()));
}
五、最佳实践总结
- 可靠性配置组合：
# 生产者
acks=all
retries=5
max.in.flight.requests.per.connection=1
# broker
unclean.leader.election.enable=false
min.insync.replicas=2
# 消费者
enable.auto.commit=false
- 监控指标要关注：
- 消息延迟监控
- 消费者lag监控
- 副本同步状态监控
- 性能优化技巧：
- 批量发送(message.size)
- 压缩消息(compression.type)
- 合理设置分区数
通过以上措施，可以构建高可靠的Kafka消息系统。实际应用中需要根据业务场景在可靠性和性能之间取得平衡。
评论