处理Kafka消息时会话超时与实现幂等性消费者

  while (true) {
     ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
     for (ConsumerRecord<String, String> record : records) {
         processMessage(record);
     }
  }

import org.apache.kafka.clients.consumer.ConsumerRecord;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.UUID; // 假设消息中包含一个业务UUID

public class IdempotentKafkaProcessor {

    private Connection dbConnection; // 数据库连接

    public IdempotentKafkaProcessor(Connection connection) {
        this.dbConnection = connection;
    }

    public void processMessage(ConsumerRecord<String, String> record) {
        String messageId = extractUniqueId(record); // 从消息中提取唯一ID，例如业务ID或Kafka生成ID

        try {
            dbConnection.setAutoCommit(false); // 开始事务

            if (isMessageAlreadyProcessed(messageId)) {
                System.out.println("消息 " + messageId + " 已处理，跳过。");
                dbConnection.rollback(); // 回滚事务，确保不提交任何更改
                return;
            }

            // 执行核心业务逻辑，例如写入数据库
            performBusinessLogic(record);

            // 标记消息为已处理
            markMessageAsProcessed(messageId);

            dbConnection.commit(); // 提交事务
            System.out.println("消息 " + messageId + " 成功处理并标记。");

        } catch (SQLException e) {
            try {
                dbConnection.rollback(); // 发生异常时回滚事务
            } catch (SQLException rollbackEx) {
                System.err.println("回滚失败: " + rollbackEx.getMessage());
            }
            System.err.println("处理消息 " + messageId + " 失败: " + e.getMessage());
            // 根据实际需求，可能需要重新抛出异常或进行其他错误处理
        } finally {
            try {
                dbConnection.setAutoCommit(true); // 恢复自动提交
            } catch (SQLException e) {
                System.err.println("恢复自动提交失败: " + e.getMessage());
            }
        }
    }

    private String extractUniqueId(ConsumerRecord<String, String> record) {
        // 实际应用中，从 record.value() 解析 JSON 或从 record.headers() 获取
        // 这里仅作示例，假设消息内容就是ID
        return record.value(); // 假设消息内容直接是唯一ID
    }

    private boolean isMessageAlreadyProcessed(String messageId) throws SQLException {
        String sql = "SELECT COUNT(*) FROM processed_messages WHERE message_id = ?";
        try (PreparedStatement ps = dbConnection.prepareStatement(sql)) {
            ps.setString(1, messageId);
            try (ResultSet rs = ps.executeQuery()) {
                if (rs.next()) {
                    return rs.getInt(1) > 0;
                }
            }
        }
        return false;
    }

    private void markMessageAsProcessed(String messageId) throws SQLException {
        String sql = "INSERT INTO processed_messages (message_id, processed_at) VALUES (?, NOW())";
        try (PreparedStatement ps = dbConnection.prepareStatement(sql)) {
            ps.setString(1, messageId);
            ps.executeUpdate();
        }
    }

    private void performBusinessLogic(ConsumerRecord<String, String> record) {
        // 实际的业务处理逻辑，例如更新用户余额、发送通知等
        System.out.println("执行业务逻辑处理消息: " + record.value());
        // 模拟业务处理耗时
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // 假设数据库表结构：
    // CREATE TABLE processed_messages (
    //     message_id VARCHAR(255) PRIMARY KEY,
    //     processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    // );
}