Kafka 中如何处理重复消费的消息？

推荐答案

在 Kafka 中处理重复消费的消息可以通过以下几种方式：

1. 幂等性生产者：确保生产者发送的消息是幂等的，即多次发送相同的消息不会导致重复的效果。Kafka 从 0.11.0 版本开始支持幂等性生产者，通过设置 enable.idempotence=true 来启用。

2. 事务性生产者：使用 Kafka 的事务 API 来确保消息的原子性发送。事务性生产者可以确保消息要么全部成功发送，要么全部失败，从而避免重复消息的产生。

3. 消费者端的去重：在消费者端实现去重逻辑，通常可以通过以下方式实现：

   • 使用外部存储：将已处理的消息 ID 存储在外部存储（如 Redis、MySQL）中，并在处理新消息时检查是否已经处理过。
   • 使用 Kafka 的偏移量管理：确保消费者在处理消息后正确提交偏移量，避免重复消费。

4. 消息的唯一标识：在消息体中包含唯一标识符（如 UUID），并在消费者端根据该标识符进行去重。

本题详细解读

1. 幂等性生产者

Kafka 的幂等性生产者通过为每个生产者实例分配一个唯一的 Producer ID（PID）和序列号（Sequence Number）来实现。当生产者发送消息时，Kafka Broker 会检查消息的序列号，如果发现重复的序列号，则会丢弃该消息，从而避免重复消息的产生。

启用幂等性生产者的方式如下：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("enable.idempotence", true);
props.put("acks", "all");
props.put("retries", 3);
props.put("max.in.flight.requests.per.connection", 1);

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

2. 事务性生产者

Kafka 的事务 API 允许生产者在一个事务中发送多条消息，并确保这些消息要么全部成功发送，要么全部失败。事务性生产者通过使用事务 ID 和事务协调器来管理事务的状态。

启用事务性生产者的方式如下：

// www.javascriptcn.com code example
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("enable.idempotence", true);
props.put("acks", "all");
props.put("retries", 3);
props.put("max.in.flight.requests.per.connection", 1);
props.put("transactional.id", "my-transactional-id");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("my-topic", "key", "value"));
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    producer.close();
} catch (KafkaException e) {
    producer.abortTransaction();
}

3. 消费者端的去重

在消费者端实现去重逻辑时，可以使用外部存储来记录已处理的消息 ID。例如，使用 Redis 来存储消息 ID：

// www.javascriptcn.com code example
Jedis jedis = new Jedis("localhost");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        String messageId = record.key();
        if (jedis.sismember("processed-messages", messageId)) {
            continue; // 跳过已处理的消息
        }
        // 处理消息
        processMessage(record.value());
        // 将消息 ID 添加到 Redis 中
        jedis.sadd("processed-messages", messageId);
    }
}

4. 消息的唯一标识

在消息体中包含唯一标识符（如 UUID），并在消费者端根据该标识符进行去重。例如：

// www.javascriptcn.com code example
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        String messageId = record.key();
        if (processedMessages.contains(messageId)) {
            continue; // 跳过已处理的消息
        }
        // 处理消息
        processMessage(record.value());
        // 将消息 ID 添加到已处理集合中
        processedMessages.add(messageId);
    }
}

通过以上方式，可以有效地处理 Kafka 中的重复消费问题。