Kafka分区中重复数据引起的乱序事件。

1. 使用消费者组：使用消费者组可以避免一个单一的消费者在处理某个分区时造成数据重复的问题。多个消费者组共同消费该主题，让每个消费者组负责一个子集分区的消费，这样每个消费者组只消费完全不同的数据。 代码示例：

consumerProps.put("group.id", "my-group"); //设置消费者组id KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps); consumer.subscribe(Collections.singletonList("my-topic"));

2. 设置消息唯一标识: 写入kafka数据时为每个消息设置唯一标识，并设置幂等性，同时禁用缓存，这样可以确保在发送过程中不会发送重复的数据。 代码示例：

producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); //设置幂等性 producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); //禁用缓存 Producer<String, String> producer = new KafkaProducer<>(producerProps); ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "my-key", "my-value"); record.headers().add(new RecordHeader("id", UUID.randomUUID().toString().getBytes())); //为每个消息设置唯一标识 producer.send(record);

3. 设置Kafka的partitioner：Kafka的Partitioner负责将消息分配到不同的分区中，我们可以实现自己的Partitioner，在将消息发送到特定分区之前通过一些策略去重。 代码示例：

public class MyPartitioner implements Partitioner { @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { //在这里实现自己的Partitioner策略 return partition; }

@Override
public void close() {}

@Override
public void configure(Map<String, ?> configs) {}

}

producerProps.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName()); Producer<String, String> producer = new KafkaProducer<>(producerProps); ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "my-key", "my-value"); producer.send(record);

以上三种方法都可以避免重复数据引起的乱序问题