Kafka - 3.x Kafka消费者不完全指北

Kafka消费模式

Kafka的consumer采用pull（拉）模式从broker中读取数据。

Kakfa消费者工作流程

消费者总体工作流程

Kafka消费者的总体工作流程包括以下步骤：

1. 配置消费者属性：首先，你需要配置消费者的属性，包括Kafka集群的地址、消费者组、主题名称、序列化/反序列化器、自动偏移提交等。
   
2. 创建消费者实例：使用配置创建Kafka消费者实例。
   
3. 订阅主题：使用消费者实例订阅一个或多个Kafka主题。这告诉Kafka消费者你想要从哪些主题中接收消息。
   
4. 轮询数据：消费者使用poll()方法从Kafka broker中拉取消息。它会定期轮询（拉）Kafka集群以获取新消息。
   
5. 处理消息：一旦从Kafka broker获取到消息，消费者会对消息进行处理，执行你的业务逻辑。这可能包括数据处理、计算、存储或其他操作。
   
6. 提交偏移量：消费者可以选择手动或自动提交偏移量，以记录已处理消息的位置。这有助于防止消息重复处理。
   
7. 处理异常：处理消息期间可能会出现异常，你需要处理这些异常，例如重试或记录错误日志。
   
8. 关闭消费者：在不再需要消费者实例时，确保关闭它以释放资源。
   

这个工作流程涵盖了Kafka消费者从配置到数据处理再到资源管理的主要步骤。消费者通常是多线程或多进程的，以处理大量的消息，并能够根据需要调整消费速率。此外，Kafka的消费者库提供了很多功能，如自动负载均衡、自动偏移管理等，以简化消费者的开发和维护。

消费者组原理

Kafka消费者组（Consumer Group）是一种机制，用于协调和管理多个消费者并共同消费一个或多个Kafka主题的消息。消费者组的工作原理如下：

1. 多个消费者：一个消费者组可以包含多个消费者实例，这些消费者实例协同工作以共同消费一个或多个主题的消息。
   
2. 订阅主题：所有消费者实例都订阅相同的Kafka主题。这意味着每个消息都会被消费者组中的一个实例处理，从而实现消息的负载均衡。
   
3. 消息分区：每个Kafka主题通常被分为多个分区，每个分区包含消息的一个子集。每个消费者实例负责消费一个或多个分区的消息。
   
4. 协调者：消费者组中的消费者实例会选择一个协调者（Coordinator）来管理组内的消费者。协调者通常是ZooKeeper或Kafka自身的一个特殊主题。
   
5. 偏移管理：协调者负责管理消费者组的偏移量（offset），这是消费者在主题分区中的当前位置。它会跟踪每个分区的消费进度，确保不会重复消费消息。
   
6. 分配分区：协调者会定期重新分配分区给消费者实例，以确保负载均衡和故障恢复。如果有新消费者加入组或有消费者离开组，协调者会重新分配分区。
   
7. 消费消息：每个消费者实例负责处理分配给它的分区中的消息。它会拉取消息，进行处理，并将偏移量提交给协调者。
   
8. 自动重平衡：如果消费者实例加入或退出消费者组，或者分区的分配发生变化，消费者组会自动进行重新平衡，以确保消息均匀分配。
   
9. 提交偏移量：消费者实例可以定期或根据需要提交已处理消息的偏移量，以便在故障时恢复消费进度。
   

通过这种方式，Kafka消费者组能够实现高可用性、负载均衡和容错，允许多个消费者并行处理消息，并根据需求动态调整分区分配。这使得消费者组成为了处理大规模流式数据的理想工具。

消费者组初始化流程

消费者组详细消费流程

Kafka消费者组的初始化流程包括一系列步骤，用于创建和配置消费者组的成员。以下是Kafka消费者组的初始化流程：

1. 引入Kafka客户端库：首先，确保你的应用程序中引入了Kafka客户端库，以便能够使用Kafka相关的类和功能。
   
2. 创建消费者配置：初始化消费者组前，需要创建一个消费者配置对象，其中包括了一些重要的属性，例如Kafka集群的地址、消费者组的ID、自动提交偏移量等。
   
3. 创建消费者实例：使用消费者配置，创建一个或多个消费者实例。每个实例代表一个消费者组中的一个成员。实例会自动注册到Kafka broker，并与协调者建立连接。
   
4. 订阅主题：通过消费者实例，使用subscribe()方法订阅一个或多个Kafka主题。这告诉Kafka你希望从哪些主题中接收消息。
   
5. 启动消费者：调用poll()方法开始轮询消息。这将启动消费者实例并开始拉取消息。消费者组中的每个成员都会独立执行这个步骤。
   
6. 消费消息：一旦消息被拉取，消费者实例会处理这些消息，执行你的业务逻辑。每个成员在自己的线程中处理消息。
   
7. 提交偏移量：消费者实例可以选择手动或自动提交已处理消息的偏移量。这有助于记录每个分区中消息的处理进度。
   
8. 处理异常：处理消息期间可能会出现异常，你需要适当地处理这些异常，例如重试消息或记录错误日志。
   
9. 关闭消费者：当不再需要消费者实例时，确保关闭它以释放资源。
   
10. 自动重平衡：如果有消费者实例加入或离开消费者组，或者分区的分配发生变化，Kafka会自动进行重新平衡，以确保消息均匀分配。
    

这个初始化流程涵盖了Kafka消费者组的基本步骤，从配置消费者组成员到消息的处理和消费。请注意，Kafka消费者组的初始化需要注意各个配置选项以及消费者组的协调过程，以确保正常运行和负载均衡。

独立消费者案例（订阅主题）

需求：创建一个独立消费者，消费artisan主题中的数据

注意：在消费者API代码中必须配置消费者组id。

package com.artisan.pc;
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.ArrayList;
import java.util.Properties;
/**
 * @author 小工匠
 * @version 1.0
 * @mark: show me the code , change the world
 */
public class CustomConsumer {
    public static void main(String[] args) {
        // 1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.126.171:9092");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 配置消费者组 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "artisan-group");
        // 3. 创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        // 4. 订阅主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("artisan");
        consumer.subscribe(topics);
        // 5. 拉取数据打印
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            // 6. 遍历并输出消费到的数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

① 在IDEA中执行消费者程序

② 服务器上中创建kafka生产者，并输入数据

③ 在IDEA中观察接收到的数据

ConsumerRecord(topic = artisan, partition = 2, leaderEpoch = 0, offset = 34, CreateTime = 1698630425187, serialized key size = -1, serialized value size = 13, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = first message)
ConsumerRecord(topic = artisan, partition = 2, leaderEpoch = 0, offset = 35, CreateTime = 1698630429909, serialized key size = -1, serialized value size = 15, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = seconde message)

消费者重要参数