深入理解Kafka消费者会话超时与幂等性处理

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.header.Header;

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

public class IdempotentKafkaConsumer {

    private final KafkaConsumer<String, String> consumer;
    private final MessageIdStore messageIdStore; // 假设这是一个处理ID存储服务

    public IdempotentKafkaConsumer(Properties props, MessageIdStore store) {
        this.consumer = new KafkaConsumer<>(props);
        this.messageIdStore = store;
        consumer.subscribe(Collections.singletonList("your_topic"));
    }

    public void startProcessing() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    String messageId = extractMessageId(record);
                    if (messageId == null) {
                        System.err.println("Skipping record due to missing message ID: " + record);
                        continue;
                    }

                    // 检查消息是否已处理
                    if (messageIdStore.isProcessed(messageId)) {
                        System.out.println("Message " + messageId + " already processed. Skipping.");
                        continue; // 跳过已处理的消息
                    }

                    // 实际处理消息
                    processMessage(record);

                    // 记录消息ID为已处理
                    messageIdStore.markAsProcessed(messageId);

                    // 提交偏移量 (此处为手动提交，推荐在批次处理完成后提交)
                    // consumer.commitSync(Collections.singletonMap(
                    //         new TopicPartition(record.topic(), record.partition()),
                    //         new OffsetAndMetadata(record.offset() + 1)
                    // ));

                } catch (Exception e) {
                    System.err.println("Error processing record " + record + ": " + e.getMessage());
                    // 错误处理策略：记录日志，将消息发送到死信队列等
                }
            }
            // 批量提交偏移量
            consumer.commitSync();
        }
    }

    private String extractMessageId(ConsumerRecord<String, String> record) {
        // 示例：从消息头中提取UUID作为消息ID
        for (Header header : record.headers()) {
            if ("message-id".equals(header.key())) {
                return new String(header.value());
            }
        }
        // 如果消息体中包含业务ID，也可以从这里提取
        // return parseBusinessIdFromJson(record.value());
        return null;
    }

    private void processMessage(ConsumerRecord<String, String> record) {
        // 模拟消息处理逻辑
        System.out.println("Processing message: " + record.value() + " with ID: " + extractMessageId(record));
        // 例如：写入数据库，调用外部API等
        try {
            Thread.sleep(50); // 模拟耗时操作
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // 假设的存储接口，用于去重
    interface MessageIdStore {
        boolean isProcessed(String messageId);
        void markAsProcessed(String messageId);
    }

    // 数据库实现示例
    static class DatabaseMessageIdStore implements MessageIdStore {
        // 实际应用中会使用数据库连接池和ORM框架
        @Override
        public boolean isProcessed(String messageId) {
            // 查询数据库，如果ID存在则返回true
            // SELECT COUNT(*) FROM processed_messages WHERE message_id = ?
            // return count > 0;
            return false; // 示例总是返回false
        }

        @Override
        public void markAsProcessed(String messageId) {
            // 插入数据库，如果因为唯一索引冲突而失败，说明已处理
            // INSERT INTO processed_messages (message_id, processed_time) VALUES (?, NOW())
            System.out.println("Marking message ID " + messageId + " as processed in DB.");
        }
    }

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my_idempotent_consumer_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"); // 禁用自动提交，手动控制
        props.put("session.timeout.ms", "10000"); // 10秒会话超时
        props.put("heartbeat.interval.ms", "3000"); // 3秒心跳间隔

        MessageIdStore store = new DatabaseMessageIdStore(); // 实际使用中会是真实的DB实现
        IdempotentKafkaConsumer consumer = new IdempotentKafkaConsumer(props, store);
        consumer.startProcessing();
    }
}