1.3.2 独立消费者案例（订阅分区）
1、需求：创建一个独立消费者，消费first主题0号分区的数据
2、实现步骤
（1）代码编写
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); } } } }
3、测试
（1）在 IDEA 中执行消费者程序。
（2）在 IDEA 中执行生产者程序 CustomProducerCallback()在控制台观察生成几个 0 号分区的数据。
（3）在 IDEA 控制台，观察接收到的数据，只能消费到 0 号分区数据表示正确。
1.3.3 消费者组案例
1、需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费
2、案例实操
（1）复制一份基础消费者的代码，在 IDEA 中同时启动，即可启动同一个消费者组中的两个消费者。
（2）启动代码中的生产者发送消息，在 IDEA 控制台即可看到两个消费者在消费不同分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码 Thread.sleep(2);）
（3）重新发送到一个全新的主题中，由于默认创建的主题分区数为 1，可以看到只能有一个消费者消费到数据。
1.4 生产经验——分区的分配以及再平衡
1、一个consumer group中有多个consumer组成，一个topic有多个partition组成，现在的问题是到底是由哪个consumer来消费哪个分区的数据。
2、kafka有四种主流的分区分配策略：Range、RoundRobin、Sticky、CooperativeSticky。可以通过配置参数partition.assignment.strategy，修改分区的分配策略。默认策略是Range+CooperativeSticky。kafka可以同时使用多个分区分配策略。
1.4.1 Range以及再平衡
1、Range分区策略原理
Range是对每个topic而言的。
首先对同一个topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。
假如现在有 7 个分区，3 个消费者，排序后的分区将会是0,1,2,3,4,5,6；消费者排序完之后将会是C0,C1,C2。
通过partition数/consumer数来决定每个消费者应该消费几个分区，如果除不尽，那么前面几个消费者将会多消费1个分区。
注意：如果只是针对一个topic而言，c0消费者多消费1个分区影响不是很大。但是如果有N多个topic，那么针对每个topic，消费者c0都将会多消费一个分区，topic越多，c0消费者的分区会比其他消费者明显多消费n个分区。
容易产生数据倾斜
2、Range分区分配策略案例
（1）修改主题first为7个分区
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 7
注意：分区数可以增加，但是不能减少
（2）复制 CustomConsumer 类，创建 CustomConsumer2。这样可以由三个消费者CustomConsumer、CustomConsumer1、CustomConsumer2 组成消费者组，组名都为“test”，同时启动 3 个消费者。
（3）启动 CustomProducer 生产者，发送 500 条消息，随机发送到不同的分区。
3、Range 分区分配再平衡案例
（1）停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。
1 号消费者：消费到4、5、6号分区数据。
2 号消费者：消费到2、3号分区数据。
0 号消费者的任务会整体被分配到 1 号消费者或者 2 号消费者。
说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。
（2）再次重新发送消息观看结果（45s 以后）。
1 号消费者：消费到 0、1、2、3 号分区数据。
2 号消费者：消费到 4、5、6 号分区数据。
说明：消费者 0 已经被踢出消费者组，所以重新按照 range 方式分配。
1.4.2 RoundRobin以及再平衡
1、RoundRobin 分区策略原理
RoundRobin针对集群中所以Topic而言。
RoundRobin轮询分区策略，是把所有的partition和所有的consumer都列出来，然后按照hashcode进行排序，最后通过轮询算法来分配partition给到各个消费者。
1.4.3 Sticky以及再平衡
粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。
粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。
1.5 offset位移
1.5.1 offset的默认维护位置
_consumer_offsets主题里面采用key和value的方式储存数据。key是group.id+topic+分区号，value值就是当前offset的值。每隔一段时间，kafka内部会对这个topic进行compact，也就是每个group.id+topic+分区号就保留最新数据。
1.5.2 自动提交offset
为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。
自动提交offset的相关参数：
enable.auto.commit:是否开启自动提交offset功能，默认是true
auto.commit.interval.ms:自动提交offset的时间间隔，默认是5s。
1.5.3 手动提交offset
虽然自动提交offset十分简单便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。
手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交；不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而异步提交则没有失败重试机制，故有可能提交失败。
commitSync(同步提交)：必须等待offset提交完毕，再去消费下一批数据。
commitAsync（异步提交）：发送完提交offset请求后，就开始消费下一批数据。
1.5.4 指定offset消费
auto.offset.reset = earliest | latest | none 默认是 latest。
当 Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？
（1）earliest：自动将偏移量重置为最早的偏移量，–from-beginning。
（2）latest（默认值）：自动将偏移量重置为最新偏移量。
（3）none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。
（4）任意指定 offset 位移开始消费
1.5.5 漏消费和重复消费
重复消费：已经消费了数据，但是 offset 没提交。
漏消费：先提交 offset 后消费，有可能会造成数据的漏消费。
1.6 消费者事务
如果想完成Consumer端的精准一次性消费，那么需要Kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质（比 如MySQL）。
1.7 数据积压（消费者如何提高吞吐量）
1、如果是Kafka消费能力不足，则可以考虑增加Topic的分区数，并且同时提升消费组的消费者数量，消费者数 = 分区数。（两者缺一不可）
2、如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压。
