JAVA 后端如何用 Kafka 实现消息幂等性，彻底避免重复消费问题 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Java 后端 Kafka 消息幂等性实现：彻底避免重复消费

大家好！今天我们来深入探讨一个在 Kafka 消息处理中至关重要的问题：幂等性。在分布式系统中，消息重复消费是常见现象，尤其是在网络波动或系统故障时。如果不对消息进行幂等性处理，可能会导致数据不一致，业务逻辑错误等严重问题。本次讲座将围绕 Java 后端如何利用 Kafka 特性以及一些常用策略，彻底避免重复消费问题。

1. 什么是幂等性

幂等性是指一个操作，无论执行多少次，产生的效果都与执行一次相同。换句话说，多次执行同一操作，不会对系统状态造成额外的改变。

例如：

幂等操作：
- 设置一个变量的值：x = 5 (无论执行多少次，x 的值最终都是 5)
- 数据库更新操作：UPDATE products SET quantity = 10 WHERE id = 123 (无论执行多少次，id 为 123 的产品的数量最终都是 10)
非幂等操作：
- 累加一个变量的值：x = x + 1 (每次执行都会改变 x 的值)
- 数据库更新操作：UPDATE products SET quantity = quantity + 1 WHERE id = 123 (每次执行都会增加产品的数量)

在消息队列的场景下，幂等性意味着即使消费者多次收到同一条消息，其业务逻辑也应该只被执行一次，并保证数据的一致性。

2. Kafka 消息传递语义

要实现幂等性，首先需要了解 Kafka 的消息传递语义。 Kafka 提供了三种消息传递语义：

最多一次（At-Most-Once）： 消息可能会丢失，但绝不会被重复消费。
最少一次（At-Least-Once）： 消息绝不会丢失，但可能会被重复消费。
精确一次（Exactly-Once）： 消息既不会丢失，也不会被重复消费。

默认情况下，Kafka 采用 最少一次 的传递语义。这意味着消费者在消费消息后，如果未及时提交 offset，或者在提交 offset 之前发生故障，则 Kafka 会认为该消息未被消费，并将其重新发送给其他消费者，从而导致重复消费。

要实现真正的幂等性，我们需要将 Kafka 的 最少一次 语义，结合业务逻辑的处理，最终达到 精确一次 的效果。

3. Kafka 实现幂等性的方案

以下介绍几种在 Java 后端使用 Kafka 实现消息幂等性的常用方案：

3.1 消息 ID 机制

这是最常见也是最基础的幂等性实现方案。为每条消息分配一个唯一的 ID，在消费者端维护一个已处理消息 ID 的集合。当消费者收到消息时，首先检查该消息 ID 是否已存在于集合中。

如果存在，则丢弃该消息，不再进行处理。
如果不存在，则进行处理，并将该消息 ID 添加到集合中。

代码示例：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class MessageIdempotentConsumer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String PROCESSED_MESSAGE_KEY_PREFIX = "processed_message:";
    private static final long MESSAGE_EXPIRY_TIME_SECONDS = 60 * 60 * 24; // 消息ID的过期时间，这里设置为24小时

    @KafkaListener(topics = "your_topic", groupId = "your_group")
    public void consume(ConsumerRecord<String, String> record) {
        String messageId = extractMessageId(record.value()); // 从消息内容中提取消息ID，假设消息体是JSON格式
        String redisKey = PROCESSED_MESSAGE_KEY_PREFIX + messageId;

        // 检查消息是否已经被处理过
        Boolean isProcessed = redisTemplate.hasKey(redisKey);
        if (Boolean.TRUE.equals(isProcessed)) {
            System.out.println("Message with ID " + messageId + " has already been processed. Ignoring.");
            return;
        }

        // 处理消息
        try {
            processMessage(record.value());
            // 将消息ID存入Redis，并设置过期时间
            redisTemplate.opsForValue().set(redisKey, "processed", MESSAGE_EXPIRY_TIME_SECONDS, TimeUnit.SECONDS);
            System.out.println("Successfully processed message with ID " + messageId);
        } catch (Exception e) {
            System.err.println("Error processing message with ID " + messageId + ": " + e.getMessage());
            // 根据业务需求，选择是否重试或记录错误日志
        }
    }

    private String extractMessageId(String message) {
        // 假设消息是JSON格式，从中提取 messageId 字段
        // 例如：{"messageId": "12345", "data": "some data"}
        try {
            // 使用 Jackson 或者 Gson 等 JSON 解析库
            com.fasterxml.jackson.databind.JsonNode jsonNode = new com.fasterxml.jackson.databind.ObjectMapper().readTree(message);
            return jsonNode.get("messageId").asText();
        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
            System.err.println("Failed to parse message as JSON: " + e.getMessage());
            return null; // 或者抛出异常，根据实际情况处理
        }
    }

    private void processMessage(String message) {
        // 这里编写你的业务逻辑，处理消息
        System.out.println("Processing message: " + message);
        // 示例：更新数据库
        // 假设消息内容包含需要更新的数据
        // productRepository.updateProduct(extractProductId(message), extractQuantity(message));
    }

    //辅助方法，用于从消息中提取productId和quantity，根据消息格式实现
    private Long extractProductId(String message) {
        try {
            com.fasterxml.jackson.databind.JsonNode jsonNode = new com.fasterxml.jackson.databind.ObjectMapper().readTree(message);
            return jsonNode.get("productId").asLong();
        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
            System.err.println("Failed to parse message as JSON: " + e.getMessage());
            return null;
        }
    }
    private Integer extractQuantity(String message) {
        try {
            com.fasterxml.jackson.databind.JsonNode jsonNode = new com.fasterxml.jackson.databind.ObjectMapper().readTree(message);
            return jsonNode.get("quantity").asInt();
        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
            System.err.println("Failed to parse message as JSON: " + e.getMessage());
            return null;
        }
    }
}

说明：

redisTemplate: 使用 Spring Data Redis 提供的 StringRedisTemplate 操作 Redis。
PROCESSED_MESSAGE_KEY_PREFIX: Redis Key 的前缀，用于区分已处理的消息 ID。
MESSAGE_EXPIRY_TIME_SECONDS: 消息 ID 在 Redis 中的过期时间，避免 Redis 占用过多内存。根据业务场景设置合理的过期时间，一般要大于消息的最大重试间隔时间。
@KafkaListener: 使用 Spring Kafka 提供的注解监听 Kafka 主题。
extractMessageId(String message): 从消息内容中提取消息 ID。这里假设消息体是 JSON 格式，你可以根据实际的消息格式进行调整。
processMessage(String message): 处理消息的业务逻辑。
Redis Key 的设计: 使用 PROCESSED_MESSAGE_KEY_PREFIX + messageId 作为 Redis Key，保证 Key 的唯一性。
Redis 的使用: 使用 Redis 的 hasKey() 方法判断消息是否已经被处理过。如果不存在，则处理消息，并将消息 ID 存入 Redis，并设置过期时间。
异常处理: 在 processMessage() 方法中，需要进行异常处理，根据业务需求选择是否重试或记录错误日志。
关于Redis: 这里的Redis可以替换为其他存储介质，比如数据库的某个字段或者BloomFilter。但是Redis作为缓存，性能较高，较为常用。

优点：

实现简单，易于理解。
适用于各种消息类型。

缺点：

需要额外的存储空间来维护已处理消息 ID 的集合。
如果消息量巨大，集合可能会非常大，影响性能。
需要考虑集合的持久化问题，防止系统重启后丢失已处理消息 ID 的信息。
如果ID过期时间过短，可能导致重复消费，如果ID过期时间过长，可能导致存储压力过大。

适用场景：

对消息处理的实时性要求不高，可以容忍一定的延迟。
消息量不是特别巨大。
可以接受额外的存储空间消耗。

3.2 数据库唯一约束

如果消息处理涉及到数据库操作，可以利用数据库的唯一约束来实现幂等性。在数据库表中创建一个唯一索引，包含消息的关键字段（例如消息 ID），当重复消费消息时，由于违反唯一约束，数据库操作会失败，从而保证了幂等性。

代码示例：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Component
public class DatabaseIdempotentConsumer {

    @Autowired
    private OrderRepository orderRepository;

    @KafkaListener(topics = "your_topic", groupId = "your_group")
    @Transactional // 保证事务性，要么全部成功，要么全部失败
    public void consume(ConsumerRecord<String, String> record) {
        String message = record.value();
        try {
            Order order = extractOrderFromMessage(message); // 从消息中提取订单信息
            orderRepository.save(order); // 保存订单到数据库
            System.out.println("Successfully processed order: " + order.getOrderId());
        } catch (DataIntegrityViolationException e) {
            // 违反唯一约束，说明订单已经被处理过
            System.out.println("Order already processed: " + message);
            // 可以记录日志，或者进行其他处理
        } catch (Exception e) {
            System.err.println("Error processing order: " + e.getMessage());
            // 根据业务需求，选择是否重试或记录错误日志
            throw e; // 重新抛出异常，让 Kafka 能够重试
        }
    }

    private Order extractOrderFromMessage(String message) {
        // 假设消息是JSON格式，从中提取订单信息
        // 例如：{"orderId": "12345", "productId": "1001", "quantity": 2}
        try {
            com.fasterxml.jackson.databind.JsonNode jsonNode = new com.fasterxml.jackson.databind.ObjectMapper().readTree(message);
            Order order = new Order();
            order.setOrderId(jsonNode.get("orderId").asText());
            order.setProductId(jsonNode.get("productId").asLong());
            order.setQuantity(jsonNode.get("quantity").asInt());
            return order;
        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
            System.err.println("Failed to parse message as JSON: " + e.getMessage());
            return null;
        }
    }
}

// Order 实体类
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;

@Entity
@Table(name = "orders")
public class Order {

    @Id
    @Column(name = "order_id")
    private String orderId;

    @Column(name = "product_id")
    private Long productId;

    @Column(name = "quantity")
    private Integer quantity;

    // Getters and setters
    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public Long getProductId() {
        return productId;
    }

    public void setProductId(Long productId) {
        this.productId = productId;
    }

    public Integer getQuantity() {
        return quantity;
    }

    public void setQuantity(Integer quantity) {
        this.quantity = quantity;
    }
}

// OrderRepository 接口
import org.springframework.data.jpa.repository.JpaRepository;

public interface OrderRepository extends JpaRepository<Order, String> {
}

SQL示例 (创建表和唯一索引):

CREATE TABLE orders (
    order_id VARCHAR(255) NOT NULL,
    product_id BIGINT,
    quantity INT,
    PRIMARY KEY (order_id)
);

-- 创建唯一索引，保证 order_id 的唯一性
-- ALTER TABLE orders ADD CONSTRAINT unique_order_id UNIQUE (order_id);  -- MySQL

说明：

@Transactional: 使用 Spring 的事务管理，保证数据的一致性。
DataIntegrityViolationException: 捕获违反唯一约束的异常，说明消息已经被处理过。
extractOrderFromMessage(String message): 从消息内容中提取订单信息，并创建 Order 对象。
数据库唯一索引: 在 orders 表上创建唯一索引，确保 order_id 的唯一性。
JPA: 使用 JPA 简化数据库操作

优点：

实现简单，利用数据库的特性。
无需额外的存储空间。

缺点：

仅适用于消息处理涉及到数据库操作的场景。
如果数据库性能瓶颈，可能会影响消息处理的效率。
依赖于数据库的事务特性。

适用场景：

消息处理涉及到数据库操作。
数据库性能可以满足消息处理的需求。
对数据一致性要求较高。

3.3 基于 Kafka 事务的幂等性

Kafka 0.11 版本之后引入了事务特性，可以实现跨分区、跨 Topic 的 精确一次 消息传递。 Kafka 事务的原理是：

生产者在发送消息前，先获取一个唯一的 transactionalId。
生产者将消息发送到 Kafka，并标记为事务的一部分。
生产者提交或中止事务。
Kafka Broker 确保事务内的所有消息要么全部成功写入，要么全部不写入。

消费者需要配置 isolation.level 参数，才能读取到已提交的事务消息。

read_uncommitted（默认值）：消费者可以读取到所有消息，包括未提交的事务消息。
read_committed：消费者只能读取到已提交的事务消息。

代码示例：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class TransactionalProducer {

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String topic = "your_topic";
        String transactionalId = "your_transactional_id"; // 必须唯一

        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 启用幂等性
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        // 设置事务 ID
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId);
        // 设置 ACK
        properties.put(ProducerConfig.ACKS_CONFIG, "all");  // 确保所有副本都写入成功

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // 初始化事务
        producer.initTransactions();

        try {
            // 开启事务
            producer.beginTransaction();

            // 发送消息
            producer.send(new ProducerRecord<>(topic, "key1", "message1"));
            producer.send(new ProducerRecord<>(topic, "key2", "message2"));

            // 模拟业务逻辑
            // ...

            // 提交事务
            producer.commitTransaction();
            System.out.println("Transaction committed successfully.");

        } catch (Exception e) {
            System.err.println("Error during transaction: " + e.getMessage());
            // 中止事务
            producer.abortTransaction();
            System.out.println("Transaction aborted.");
        } finally {
            producer.close();
        }
    }
}

import org.apache.kafka.clients.consumer.ConsumerConfig;
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 TransactionalConsumer {

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String topic = "your_topic";
        String groupId = "your_group";

        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 设置隔离级别，只能读取已提交的事务消息
        properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        // 禁用自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        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());
                    // 处理消息
                    // ...
                }
                // 手动提交 offset
                consumer.commitSync();
            }
        } catch (Exception e) {
            System.err.println("Error during consumption: " + e.getMessage());
        } finally {
            consumer.close();
        }
    }
}

说明：

Producer:
- ENABLE_IDEMPOTENCE_CONFIG: 启用幂等性，Kafka 会自动处理消息的重复发送。
- TRANSACTIONAL_ID_CONFIG: 设置事务 ID，必须全局唯一。
- ACKS_CONFIG: 设置为 "all"，确保所有副本都写入成功。
- initTransactions(): 初始化事务。
- beginTransaction(): 开启事务。
- commitTransaction(): 提交事务。
- abortTransaction(): 中止事务。
Consumer:
- ISOLATION_LEVEL_CONFIG: 设置为 read_committed，确保只能读取到已提交的事务消息。
- ENABLE_AUTO_COMMIT_CONFIG: 设置为 false，禁用自动提交 offset，需要手动提交 offset。
- commitSync(): 手动同步提交 offset。

优点：

真正的 精确一次 消息传递语义。
可以实现跨分区、跨 Topic 的事务。

缺点：

实现复杂，需要理解 Kafka 事务的原理。
性能开销较大，会降低吞吐量。
需要 Kafka Broker 的支持（0.11 版本及以上）。

适用场景：

对数据一致性要求极高。
可以容忍一定的性能损失。
需要跨分区、跨 Topic 的事务。

3.4 基于业务逻辑的幂等性

除了使用 Kafka 提供的特性外，还可以从业务逻辑层面来实现幂等性。例如，对于更新操作，可以将更新操作转换为设置操作。对于一些特殊场景，可以利用状态机来实现幂等性。

示例：

假设有一个订单状态更新的场景，订单状态有：CREATED、PAID、SHIPPED、COMPLETED。可以通过状态机来控制订单状态的流转，保证每个状态只能被更新一次。

代码示例：

public enum OrderStatus {
    CREATED,
    PAID,
    SHIPPED,
    COMPLETED
}

public class Order {
    private String orderId;
    private OrderStatus status;

    public Order(String orderId) {
        this.orderId = orderId;
        this.status = OrderStatus.CREATED;
    }

    public String getOrderId() {
        return orderId;
    }

    public OrderStatus getStatus() {
        return status;
    }

    public void pay() {
        if (this.status == OrderStatus.CREATED) {
            this.status = OrderStatus.PAID;
            System.out.println("Order " + orderId + " paid successfully.");
        } else {
            System.out.println("Order " + orderId + " cannot be paid. Current status: " + status);
        }
    }

    public void ship() {
        if (this.status == OrderStatus.PAID) {
            this.status = OrderStatus.SHIPPED;
            System.out.println("Order " + orderId + " shipped successfully.");
        } else {
            System.out.println("Order " + orderId + " cannot be shipped. Current status: " + status);
        }
    }

    public void complete() {
        if (this.status == OrderStatus.SHIPPED) {
            this.status = OrderStatus.COMPLETED;
            System.out.println("Order " + orderId + " completed successfully.");
        } else {
            System.out.println("Order " + orderId + " cannot be completed. Current status: " + status);
        }
    }

    public static void main(String[] args) {
        Order order = new Order("12345");
        order.pay();
        order.ship();
        order.complete();

        // 模拟重复消费
        order.pay();
        order.ship();
        order.complete();
    }
}

说明：

OrderStatus 枚举定义了订单的状态。
Order 类维护了订单的状态，并提供了状态更新的方法。
每个状态更新方法都会检查当前状态是否允许更新，从而保证了幂等性。

优点：

无需依赖 Kafka 的特性。
可以根据业务逻辑灵活实现。

缺点：

实现复杂，需要深入理解业务逻辑。
不适用于所有场景。

适用场景：

可以从业务逻辑层面实现幂等性。
对性能要求较高。

4. 方案选择

选择哪种方案取决于具体的业务场景。

方案	优点	缺点	适用场景
消息 ID 机制	实现简单，易于理解，适用于各种消息类型。	需要额外的存储空间来维护已处理消息 ID 的集合，如果消息量巨大，集合可能会非常大，影响性能，需要考虑集合的持久化问题，防止系统重启后丢失已处理消息 ID 的信息。	对消息处理的实时性要求不高，可以容忍一定的延迟，消息量不是特别巨大，可以接受额外的存储空间消耗。
数据库唯一约束	实现简单，利用数据库的特性，无需额外的存储空间。	仅适用于消息处理涉及到数据库操作的场景，如果数据库性能瓶颈，可能会影响消息处理的效率，依赖于数据库的事务特性。	消息处理涉及到数据库操作，数据库性能可以满足消息处理的需求，对数据一致性要求较高。
基于 Kafka 事务的幂等性	真正的精确一次消息传递语义，可以实现跨分区、跨 Topic 的事务。	实现复杂，需要理解 Kafka 事务的原理，性能开销较大，会降低吞吐量，需要 Kafka Broker 的支持（0.11 版本及以上）。	对数据一致性要求极高，可以容忍一定的性能损失，需要跨分区、跨 Topic 的事务。
基于业务逻辑的幂等性	无需依赖 Kafka 的特性，可以根据业务逻辑灵活实现。	实现复杂，需要深入理解业务逻辑，不适用于所有场景。	可以从业务逻辑层面实现幂等性，对性能要求较高。

5. 注意事项

消息 ID 的选择: 消息 ID 必须具有全局唯一性，可以使用 UUID 或 Snowflake 算法生成。
存储介质的选择: 选择合适的存储介质来存储已处理的消息 ID，例如 Redis、数据库、Bloom Filter 等。
过期时间的设置: 为已处理的消息 ID 设置合理的过期时间，避免存储介质占用过多空间。
异常处理: 在消息处理过程中，需要进行异常处理，保证消息的可靠性。
监控和告警: 对消息处理过程进行监控，及时发现和处理异常情况。
offset提交: 记得手动提交offset，避免重复消费。

6. 实现消息幂等性的一些思考

实现消息幂等性不仅仅是技术上的问题，更需要结合业务场景进行思考。在设计系统时，应该尽量避免对消息顺序有严格要求的场景。如果必须保证消息的顺序性，可以考虑将相关的消息发送到同一个分区，并使用单线程的消费者进行处理。此外，还需要考虑消息的重试机制，避免因为消息处理失败而导致数据不一致。

7. 避免重复消费，保障数据一致

本次讲座主要介绍了在 Java 后端如何利用 Kafka 特性以及一些常用策略，来实现消息的幂等性，从而彻底避免重复消费的问题。通过消息ID机制、数据库唯一约束、Kafka事务以及基于业务逻辑的方法，结合实际场景选择合适的方案，可以保障数据的一致性，构建稳定可靠的分布式系统。

Java 后端 Kafka 消息幂等性实现：彻底避免重复消费

1. 什么是幂等性

2. Kafka 消息传递语义

3. Kafka 实现幂等性的方案

3.1 消息 ID 机制

3.2 数据库唯一约束

3.3 基于 Kafka 事务的幂等性

3.4 基于业务逻辑的幂等性

4. 方案选择

5. 注意事项

6. 实现消息幂等性的一些思考

7. 避免重复消费，保障数据一致

发表回复 取消回复

发表回复取消回复