Kafka面试题----如何保证Kafka消费者在消费过程中不丢失消息

合理配置消费者参数

enable.auto.commit：设置为 false，关闭自动提交偏移量。自动提交偏移量存在一定的时间间隔，在这个间隔内如果消费者出现异常，可能会导致部分消息被重复消费或者丢失。关闭自动提交后，由开发者手动控制偏移量的提交，确保消息处理完成后再提交偏移量。

enable.auto.commit=false

auto.offset.reset：根据业务需求合理设置该参数。该参数指定当消费者没有有效的偏移量时（例如，消费者首次启动或者偏移量已过期），从何处开始消费消息。
- earliest：从分区的起始位置开始消费消息，确保不会遗漏任何消息，但可能会导致重复消费。
- latest：从分区的最新位置开始消费消息，会忽略分区中已有的旧消息。如果需要处理所有消息，不建议使用该值。

auto.offset.reset=earliest

正确处理消息和提交偏移量

手动提交偏移量：在消息处理完成后，手动提交偏移量。可以选择同步提交（commitSync()）或异步提交（commitAsync()）。

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class ManualOffsetCommitConsumer {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "test-group");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("enable.auto.commit", "false");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);String topic = "test-topic";consumer.subscribe(Collections.singletonList(topic));try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {// 处理消息System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());}// 同步提交偏移量consumer.commitSync();}} finally {consumer.close();}}
}

批量处理和批量提交：如果消息处理速度较慢，可以采用批量处理的方式，处理完一批消息后再统一提交偏移量。但要注意，批量处理过程中如果出现异常，可能需要回滚并重新处理这批消息。

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;public class BatchOffsetCommitConsumer {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "test-group");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("enable.auto.commit", "false");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);String topic = "test-topic";consumer.subscribe(Collections.singletonList(topic));try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();for (ConsumerRecord<String, String> record : records) {// 处理消息System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());TopicPartition partition = new TopicPartition(record.topic(), record.partition());// 记录每个分区的最新偏移量offsetsToCommit.put(partition, new OffsetAndMetadata(record.offset() + 1));}if (!offsetsToCommit.isEmpty()) {// 同步提交偏移量consumer.commitSync(offsetsToCommit);}}} finally {consumer.close();}}
}

处理消费者异常和故障恢复

异常处理：在消息处理过程中，捕获并处理可能出现的异常，确保在异常情况下不会丢失消息。例如，如果处理消息时出现数据库写入失败的异常，可以进行重试或者记录日志，待问题解决后重新处理。

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class ExceptionHandlingConsumer {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "test-group");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("enable.auto.commit", "false");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);String topic = "test-topic";consumer.subscribe(Collections.singletonList(topic));try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {try {// 处理消息System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());} catch (Exception e) {// 记录异常日志System.err.println("Error processing message: " + e.getMessage());// 可以进行重试或者其他处理}}consumer.commitSync();}} finally {consumer.close();}}
}