51工具盒子嗨，你好呀，我是猿java
在 Apache Kafka 中，避免重复消费是一个常见的问题，尤其是在处理消息时需要确保每条消息只被处理一次。那么，有什么方式可以避免重复消费？这篇文章，我们来聊一聊。
通常来说，避免重复消费的方式有 7种：
- 使用消费者组 {#1-使用消费者组} =====================
Kafka的消费者组（Consumer Group）机制可以确保每个分区的消息只被一个消费者实例消费。通过合理的分区和消费者组设计，可以避免同一消息被多个消费者重复消费。
优点：
- 简单易用，Kafka内置支持。
- 适用于简单的负载均衡和扩展。
缺点：
- 不能完全避免重复消费，比如在消费者重启或重新平衡的过程中可能会有些消息被重复消费。
- 需要额外处理消费者重平衡带来的复杂性。
- 使用幂等生产者 {#2-使用幂等生产者} =======================
Kafka 0.11.0版本引入了幂等生产者（Idempotent Producer），可以确保相同的消息在网络或其他错误导致重试时不会被重复写入Kafka。启用幂等生产者只需要在生产者配置中设置enable.idempotence=true。幂等生产者确保消息在网络或其他错误导致重试时不会被重复写入 Kafka，通过为每个消息分配唯一的序列号来实现幂等性。
配置修改如下：
|-----------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| 1 2 3 4 | Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("enable.idempotence", "true"); KafkaProducer<String, String> producer = new KafkaProducer<>(props); |
优点：
- 简化了生产者端的去重逻辑。
- 可以确保消息在Kafka中只写入一次。
缺点：
- 需要Kafka 0.11.0及以上版本。
- 在某些情况下可能会增加生产者的延迟。
- 使用事务性生产者和消费者 {#3-使用事务性生产者和消费者} =================================
Kafka支持事务性消息，允许生产者和消费者在一个事务中一起工作。生产者可以将一组消息作为一个事务写入Kafka，消费者也可以在一个事务中读取和处理消息。这样可以确保消息处理的原子性和一致性。要使用事务性生产者，需要配置transactional.id。
配置修改如下：
|-------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| 1 2 3 4 5 | Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("transactional.id", "my-transactional-id"); KafkaProducer<String, String> producer = new KafkaProducer<>(props); producer.initTransactions(); |
优点：
- 提供了强一致性保证。
- 避免了消息处理中的部分提交问题。
缺点：
- 复杂度较高，需Kafka 0.11.0及以上版本。
- 性能开销较大，适用于对一致性要求高的场景。
- 手动提交偏移量 {#4-手动提交偏移量} =======================
默认情况下，Kafka消费者会自动提交偏移量（auto commit），为了更好地控制消息处理和偏移量提交，可以关闭自动提交（enable.auto.commit=false），并在确保消息处理成功后手动提交偏移量。这可以通过commitSync()或commitAsync()方法来实现。
配置修改如下：
|------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| 1 2 3 4 5 6 7 8 9 10 11 12 | Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("enable.auto.commit", "false"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { // 处理消息 } consumer.commitSync(); } |
优点：
- 精细控制偏移量提交时机，确保消息处理成功后才提交。
- 提高了处理的可靠性。
缺点：
- 增加了消费者代码的复杂性。
- 如果处理逻辑很慢，可能导致偏移量提交延迟。
- 使用外部存储来管理偏移量 {#5-使用外部存储来管理偏移量} =================================
在某些场景下，可以将偏移量存储在外部存储（如数据库）中，而不是依赖 Kafka的内部偏移量管理。这样可以在消息处理和偏移量提交之间建立更强的关联，确保只有当消息处理成功后才更新偏移量。
优点：
- 可以在消息处理和偏移量提交之间建立更强的关联。
- 灵活性高，可以根据业务需求自定义偏移量管理。
缺点：
- 需要额外的存储和管理逻辑。
- 增加了系统的复杂性。
- 去重逻辑 {#6-去重逻辑} =================
在消息处理逻辑中引入去重机制。例如，可以使用消息的唯一标识符（如消息ID）在处理前检查是否已经处理过该消息，从而避免重复处理。
优点：
- 灵活性高，可以根据业务逻辑自定义去重策略。
- 适用于需要严格去重的场景。
缺点：
- 需要额外的存储和管理去重信息。
- 增加了处理逻辑的复杂性。
- 幂等的消息处理逻辑 {#7-幂等的消息处理逻辑} ===========================
设计消息处理逻辑时，尽量使其成为幂等操作，即相同的消息即使被处理多次也不会产生副作用。
例如，在数据库操作时，可以使用UPSERT操作（更新插入）来确保数据的一致性。
优点：
- 简化了重复消费问题的处理。
- 适用于可以设计为幂等操作的业务场景。
缺点：
- 并不是所有业务逻辑都能设计为幂等操作。
- 需要仔细设计和验证处理逻辑的幂等性。
总结 {#总结}
本文分析了在 Kafka 中，避免重复消费的 7种常见方式，对于大多数场景，结合使用消费者组、手动提交偏移量和幂等处理逻辑可以有效避免重复消费，而在需要更严格一致性的场景下，可以考虑使用幂等生产者和事务性消息。具体选择哪种方法取决于具体的应用场景和需求。
交流学习 {#交流学习}
最后，把猿哥的座右铭送给你：投资自己才是最大的财富。 如果你觉得文章有帮助，请帮忙转发给更多的好友，或关注公众号：猿java，持续输出硬核文章。
