【Kafka】Kafka消费者

文章目录

【Kafka】Kafka消费者
- 1. 消费方式
- - 1.1 消费者工作流程
  - 1.2 消费者组原理
  - 1.3 消费者组初始化流程
  - 1.4 消费者组详细消费流程
  - 1.5 消费者重要参数
- 2. 消费者API
- - 2.1 独立消费者案例
  - 2.2 订阅分区
  - 2.3 消费者组案例

1. 消费方式

pull(拉)模式：consumer采用从broker中主动拉取数据。Kafka采用这种方式。
push(推)模式：Kafka没有采用这种方式，因为由broker决定消息发送速率，很难适应所有的消费者的消费速率。例如推送的速度是50m/s，consumer1和consumer2旧来不及处理消息。

pull模式不足之处是，如果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据。

1.1 消费者工作流程

消费者和消费者之间是相互独立的，一个消费者可以消费多个分区数据。消费者A消费了0号分区数据消费者B也可以消费0号分区数据。
每个分区的数据只能由消费者组中多个消费者其中一个消费者消费。
消费者通过记录在系统主题的offset知道消息消费到哪一步了

1.2 消费者组原理

Consumer Group（CG）：消费者组，由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同。

消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。
消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者是逻辑上的一个订阅者。

1.3 消费者组初始化流程

coordinator：辅助实现消费者组的初始化和分区的分配。

coordinator节点选择=groupid的hashcode值 % 50(50是_consumer_offsets的分区数量)

例如：groupid的hashcode值=1，1%50=1，那么_consumer_offsets主题的1号分区，在哪个broker上，就选择这个节点的coordinator作为这个消费者组的老大。消费者组下的所有消费者提交offset的时候就往这个分区去提交offset。

1.4 消费者组详细消费流程

1.5 消费者重要参数

参数名称	描述
bootstrap.servers	向 Kafka 集群建立初始连接用到的 host/port 列表。
key.deserializer 和 value.deserializer	指定接收消息的 key 和 value 的反序列化类型。一定要写全类名。
group.id	标记消费者所属的消费者组。
enable.auto.commit	默认值为 true，消费者会自动周期性地向服务器提交偏移量。
auto.commit.interval.ms	如果设置了 enable.auto.commit 的值为 true，则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s。
auto.offset.reset	当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？ earliest：自动重置偏移量到最早的偏移量。 latest：默认，自动重置偏移量为最新的偏移量。 none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
offsets.topic.num.partitions	__consumer_offsets 的分区数，默认是 50 个分区。
heartbeat.interval.ms	Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms ，也不应该高于 session.timeout.ms 的 1/3。
session.timeout.ms	Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡。
max.poll.interval.ms	消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。
fetch.min.bytes	默认 1 个字节。消费者获取服务器端一批消息最小的字节数。
fetch.max.wait.ms	默认 500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据。
fetch.max.bytes	默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。
max.poll.records	一次 poll 拉取数据返回消息的最大条数，默认是 500 条。

2. 消费者API

2.1 独立消费者案例

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

**注意：在消费者API代码中必须配置消费者组id。**命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id

示例代码：

public class CustomConsumer {
    public static void main(String[] args) {
        //0.配置
        Properties properties = new Properties();
        //连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.66:9092,192.168.101.67:9092,192.168.101.68:9092");
        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

        //1.创建一个消费者 "","hello"
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        //2.订阅主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //3.消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }

    }
}

2.2 订阅分区

需求：创建一个独立消费者，消费first主题0号分区的数据。

代码示例如下：

public class CustomConsumerPartition {
    public static void main(String[] args) {
        //0.配置
        Properties properties = new Properties();
        //连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.66:9092,192.168.101.67:9092,192.168.101.68:9092");
        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

        //1.创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        //2.订阅主题first
        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition("first", 0));
        kafkaConsumer.assign(topicPartitions);

        //3.消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }

    }
}

2.3 消费者组案例

需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费。

复制 2.1 中类，创建两个CustomConsumer1和CustomConsumer2。这样就有了由三个消费者组成的消费者组。我们利用如下生产者代码发送数据：

public class CustomProducerCallback {
    public static void main(String[] args) {
        //0.配置
        Properties properties = new Properties();
        //连接集群 bootstrap.server
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.66:9092,192.168.101.67:9092,192.168.101.68:9092");
        //指定对应的 key 和 value 的序列化类型(key.serializer,)
        //properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //1.创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        //2.发送数据
        for (int i = 0; i < 500; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "hello,world" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("主题：" + recordMetadata.topic() + "\t分区" + recordMetadata.partition());
                    }
                }
            });
        }

        //3.关闭资源
        kafkaProducer.close();
    }
}