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

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.errors.WakeupException;

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

public class IdempotentKafkaConsumer {

    private final Consumer<String, String> consumer;
    private volatile boolean running = true;

    public IdempotentKafkaConsumer(Consumer<String, String> consumer, String topic) {
        this.consumer = consumer;
        this.consumer.subscribe(Collections.singletonList(topic));
    }

    public void run() {
        try {
            while (running) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String messageId = extractUniqueId(record); // 步骤1: 从消息中提取唯一ID

                    // 步骤2: 检查消息是否已处理
                    if (isMessageProcessed(messageId)) {
                        System.out.println("Message with ID " + messageId + " already processed. Skipping.");
                        continue; // 已处理，跳过当前消息
                    }

                    try {
                        // 步骤3: 实际处理消息，并确保操作的原子性
                        processMessage(record);
                        markMessageAsProcessed(messageId); // 标记为已处理
                        System.out.println("Processed message: " + record.offset() + " with ID: " + messageId);
                    } catch (Exception e) {
                        System.err.println("Error processing message " + messageId + ": " + e.getMessage());
                        // 根据业务需求处理异常，可能需要重试或记录失败
                    }
                }
                consumer.commitSync(); // 提交偏移量
            }
        } catch (WakeupException e) {
            // 消费者被中断，通常用于优雅关闭
            System.out.println("Consumer shutting down.");
        } finally {
            consumer.close();
        }
    }

    public void shutdown() {
        running = false;
        consumer.wakeup(); // 唤醒消费者以中断poll方法
    }

    // --- 辅助方法（需要根据实际业务逻辑实现） ---

    /**
     * 从Kafka消息中提取唯一的业务ID。
     * 这可以是消息体中的一个字段，或者是一个自定义的消息头。
     */
    private String extractUniqueId(ConsumerRecord<String, String> record) {
        // 示例：假设消息内容是JSON，包含一个"id"字段
        // 实际应用中可能需要更复杂的解析或从消息头获取
        return "business-id-" + record.value().hashCode(); // 仅作示例，实际应提取有意义的唯一ID
    }

    /**
     * 检查给定ID的消息是否已经处理过。
     * 这通常涉及查询数据库或分布式缓存。
     * 返回true表示已处理，false表示未处理。
     */
    private boolean isMessageProcessed(String messageId) {
        // 示例：查询数据库或缓存，检查是否存在该messageId的记录
        // 实际实现需要考虑并发和持久化
        return false; // 模拟未处理
    }

    /**
     * 处理消息的实际业务逻辑。
     * 这可能涉及写入数据库、调用外部API等。
     */
    private void processMessage(ConsumerRecord<String, String> record) {
        // 模拟耗时操作
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        // 实际的业务处理逻辑
    }

    /**
     * 标记给定ID的消息为已处理。
     * 这通常涉及在数据库或分布式缓存中记录该messageId。
     * 需与processMessage在同一个事务中，或通过其他机制保证原子性。
     */
    private void markMessageAsProcessed(String messageId) {
        // 示例：在数据库中插入或更新一条记录，表示该messageId已处理
        // 实际实现需要考虑事务和持久化
    }
}