进击消息中间件系列（六）：Kafka 消费者Consumer

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 条。

package org.zhm.consumer;

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;

/**
 * @ClassName CustomConsumer
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/14 9:08
 * @Version 1.0
 */
public class CustomConsumer {
    public static void main(String[] args) {
        //1、创建消费者的配置对象
        Properties properties=new Properties();

        //2、给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102: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,"test2");

//        //修改分区分配策略
//        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");

        //测试：修改分区分配策略
        ArrayList<String> startegys=new ArrayList<>();
        startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,startegys);


        //3、创建消费者对象
        KafkaConsumer<String,String> kafkaConsumer=new KafkaConsumer<String, String>(properties);

        //注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics=new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //拉取数据打印
        while (true){
            //设置1s中消费一批数据
            ConsumerRecords<String,String> consumerRecords=kafkaConsumer.poll(Duration.ofSeconds(1));

            //打印消费到的数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }

    }
}

package org.zhm.consumer;

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.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import javax.lang.model.type.ArrayType;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

/**
 * @ClassName CustomConsumerPartition
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/14 9:18
 * @Version 1.0
 */
public class CustomConsumerPartition {
    public static void main(String[] args) {
        Properties properties=new Properties();

        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102: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,"test");

        KafkaConsumer<String,String> kafkaConsumer=new KafkaConsumer<String, String>(properties);

        //消费某个主题的某个分区数据
        ArrayList<TopicPartition> topicPartitions=new ArrayList<>();

        topicPartitions.add(new TopicPartition("first",0));

        kafkaConsumer.assign(topicPartitions);

        while (true){
            ConsumerRecords<String,String> consumerRecords=kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

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 分钟。超过该值，该消费者被移除，消费者组执行再平衡。

partition.assignment.strategy
#消费者分区分配策略，默认策略是 Range +CooperativeSticky。Kafka 可以同时使用多个分区分配策略。可以选择的策略包括：Range、RoundRobin、Sticky、CooperativeStick

 bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 7