上一篇地址：持续总结中！2024年面试必问 20 道 Kafka面试题（六）-CSDN博客

十三、Kafka的消费者如何订阅和消费Topic？

在Kafka中，消费者通过消费者组（Consumer Group）来订阅和消费Topic。以下是消费者订阅和消费Topic的详细步骤：

1. 创建消费者配置：

   • 消费者首先需要创建一个配置对象，设置必要的参数，如bootstrap.servers（Kafka集群的地址列表）、group.id（消费者组的ID）、key.deserializer和value.deserializer（用于反序列化消息键和消息值的类）等。

2. 创建消费者实例：

   • 使用配置对象，创建一个KafkaConsumer实例。

3. 订阅主题：

   • 消费者使用subscribe()方法订阅一个或多个主题。订阅后，消费者组内的消费者将协调彼此，以确保负载均衡，每个分区只被组内的一个消费者消费。

4. 轮询消息：

   • 消费者通过调用poll()方法来轮询消息。poll()方法有两个参数：超时时间（以毫秒为单位），以及可选的回调函数。超时时间指定了消费者在没有消息时等待的时长。

5. 获取记录：

   • poll()方法返回一个ConsumerRecords对象，其中包含从订阅主题的分区接收到的消息记录。每个记录包含消息的键、值、分区信息和偏移量。

6. 处理消息：

   • 消费者迭代ConsumerRecords对象，对每条消息进行处理。这通常涉及反序列化消息的键和值，以及执行业务逻辑。

7. 提交偏移量：

   • Kafka消费者使用偏移量（Offset）来跟踪在每个分区中读取的位置。消费者在处理完一批消息后，可以手动提交偏移量，告诉Kafka已经成功处理了这些消息。提交偏移量可以通过commitSync()（同步提交）或commitAsync()（异步提交）方法完成。

8. 自动提交偏移量：

   • 如果配置了enable.auto.commit为true，消费者将在每次poll()调用后自动提交偏移量。自动提交可以设置为同步或异步，通过auto.commit.interval.ms参数控制提交间隔。

9. 消费者协调：

   • 在消费者组内，Kafka使用ZooKeeper或Raft（在较新版本中）来协调消费者之间的分区分配。这确保了每个分区只被组内的一个消费者消费，并且可以在消费者加入或退出组时动态地重新分配分区。

10. 消费者重平衡：

    • 当消费者组的订阅、组成员或配置发生变化时，可能会触发消费者重平衡。在重平衡期间，Kafka将重新分配分区给消费者组中的成员。

11. 关闭消费者：

    • 当消费者不再需要时，应当关闭消费者以释放资源。关闭消费者会确保所有正在进行的操作完成，并释放网络连接等资源。

12. 异常处理：

    • 在整个消费过程中，消费者需要处理可能出现的异常，例如网络问题、序列化问题等。异常处理机制可以确保消费者在遇到错误时能够做出适当的响应。

通过这些步骤，Kafka消费者可以有效地订阅和消费Topic中的消息。消费者组和分区的概念使得Kafka能够支持大规模分布式系统的消息消费需求，同时提供了高吞吐量、可靠性和灵活性。

十四、Kafka的消费者如何处理消息的重复读取？

在Kafka中，消息的重复读取通常发生在以下几种情况：

1. 消费者故障：

   • 如果消费者在处理完消息并提交偏移量之后发生故障，然后重新启动，它可能会从上次提交的偏移量开始重新读取消息。

2. 消费者组重平衡：

   • 当消费者组中的成员发生变化时（例如，新消费者加入或现有消费者退出），Kafka会触发消费者组重平衡。在重平衡过程中，分区的所有权可能会在消费者之间转移，导致消息被重新读取。

3. 手动干预：

   • 管理员或应用程序可能会手动重置消费者的偏移量，导致消费者从旧的偏移量重新读取消息。

为了处理消息的重复读取问题，Kafka提供了以下几种策略：

1. 幂等性消费者：

   • Kafka允许消费者实现幂等性，这意味着即使消息被重复读取，应用程序的状态也不会受到影响。这要求应用程序能够处理重复的消息而不产生副作用。

2. 消息去重：

   • 应用程序可以在处理消息之前检查消息的唯一标识符（如消息ID或业务唯一键），如果消息已经被处理过，则可以跳过重复的消息。

3. 使用事务：

   • 如果Kafka消费者是事务性的（即设置了transactional.id），它可以在一个事务中读取和处理消息。这样可以确保在事务提交之前，消息不会被其他消费者读取。

4. 偏移量管理：

   • 消费者可以在处理完每条消息后立即提交偏移量，这样即使发生故障，也不会导致消息的重复读取。但是，这可能会降低吞吐量。

5. 消费者组管理：

   • 通过合理设计消费者组和分区的数量，以及管理消费者的启动和停止，可以减少消费者组重平衡的发生，从而减少重复读取的可能性。

6. 消费者状态存储：

   • 消费者可以在外部存储（如数据库）中记录已经处理过的消息的状态。在处理新消息之前，消费者可以检查外部存储以确定消息是否已经被处理过。

7. 业务逻辑设计：

   • 在设计业务逻辑时，可以考虑消息的重复读取问题，并设计相应的逻辑来避免或减少重复处理的影响。

8. 使用Exactly-Once Semantics (EOS)：

   • Kafka提供了端到端的精确一次（Exactly-Once Semantics）处理语义，确保消息在生产者、消费者和存储系统中只被处理一次。这需要生产者、消费者和Kafka集群都支持EOS。

通过这些策略，Kafka消费者可以有效地处理消息的重复读取问题，确保数据的一致性和应用程序的健壮性。然而，需要注意的是，完全避免重复读取可能需要应用程序级别的支持和合理的系统设计。