提升Kafka消费者健壮性：会话超时处理与消息处理语义

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class IdempotentKafkaConsumer {

    private final Consumer<String, String> consumer;
    private final MessageProcessor messageProcessor; // 业务消息处理器

    public IdempotentKafkaConsumer(String bootstrapServers, String groupId, String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("enable.auto.commit", "false"); // 禁用自动提交，手动控制提交
        props.put("session.timeout.ms", "10000"); // 示例：会话超时时间
        props.put("heartbeat.interval.ms", "3000"); // 示例：心跳间隔

        this.consumer = new KafkaConsumer<>(props);
        this.consumer.subscribe(Collections.singletonList(topic));
        this.messageProcessor = new MessageProcessor(); // 实例化业务处理器
    }

    public void startProcessing() {
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    continue;
                }

                for (ConsumerRecord<String, String> record : records) {
                    // 获取消息的唯一标识符，例如从消息值中解析或从头部获取
                    String uniqueId = extractUniqueId(record); 

                    if (messageProcessor.isProcessed(uniqueId)) {
                        System.out.println("Message with ID " + uniqueId + " already processed. Skipping.");
                        continue; // 跳过已处理的消息
                    }

                    try {
                        messageProcessor.process(record); // 实际业务处理
                        messageProcessor.markAsProcessed(uniqueId); // 标记为已处理
                        System.out.println("Processed record: " + record.offset() + " for partition " + record.partition());
                    } catch (Exception e) {
                        System.err.println("Error processing record " + uniqueId + ": " + e.getMessage());
                        // 根据业务需求处理异常，例如记录日志、发送告警、死信队列等
                        // 不提交偏移量，以便下次重新处理
                    }
                }
                // 批次处理完成后，手动提交偏移量
                consumer.commitSync(); 
            }
        } catch (Exception e) {
            System.err.println("Consumer loop interrupted: " + e.getMessage());
        } finally {
            consumer.close();
        }
    }

    private String extractUniqueId(ConsumerRecord<String, String> record) {
        // 示例：假设消息值是JSON，包含一个"id"字段
        // 实际情况可能需要更复杂的解析或从record.headers()中获取
        return record.value().split(":")[0]; // 简化示例
    }

    // 模拟业务消息处理器
    static class MessageProcessor {
        // 存储已处理消息ID的持久化层（例如：数据库、Redis）
        // 生产环境应使用真正的持久化存储
        private final java.util.Set<String> processedIds = Collections.synchronizedSet(new java.util.HashSet<>());

        public boolean isProcessed(String uniqueId) {
            // 实际中应查询数据库或Redis
            return processedIds.contains(uniqueId);
        }

        public void process(ConsumerRecord<String, String> record) throws InterruptedException {
            // 模拟耗时业务处理
            System.out.println("Processing message: " + record.value());
            Thread.sleep(50); // 模拟处理时间
        }

        public void markAsProcessed(String uniqueId) {
            // 实际中应将ID写入数据库或Redis
            processedIds.add(uniqueId);
        }
    }

    public static void main(String[] args) {
        // 替换为您的Kafka集群地址、消费者组ID和主题
        IdempotentKafkaConsumer consumer = new IdempotentKafkaConsumer("localhost:9092", "my-group", "my-topic");
        consumer.startProcessing();
    }
}