Kafka

消息分层

消息丢失

生产者拦截器

消费者拦截器

Consumer Group 下可以有一个或多个 Consumer 实例

Group ID 是一个字符串，在一个 Kafka 集群中，它标识唯一的一个 Consumer Group

Consumer Group 下所有实例订阅的主题的单个分区，只能分配给组内的某个 Consumer 实例消费。这个分区当然也可以被其他的 Group 消费。

Rebalance

一台服务器就是一个broker

控制器故障转移（Failover）<br>

位移主题（__consumer_offsets）

每个分区是一个独立的、有序的消息日志。每条消息都带有一个偏移量（offset），表示它在分区中的位置

作用

分配策略

Partition 和 Consumer的关系<br>

分区数量

副本的优势

领导者（Leader-based）的副本机制（保障副本数据一致性）<br>

// 2 发送数据 for (int i = 0; i &lt; 500; i++) { kafkaProducer.send(new ProducerRecord&lt;&gt;(&quot;first&quot;, &quot;atguigu&quot; + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null){ System.out.println(&quot;主题： &quot;+metadata.topic() + &quot; 分区： &quot;+ metadata.partition()); } } }); Thread.sleep(2); }

分支主题

三种分区方式

1. 设置缓冲区大小

2. 设置批次大小

3. 减少linger时间

4. 设置压缩

// 配置<br> Properties properties = new Properties();<br> // 连接kafka集群<br> properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");<br> // 序列化<br> properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());<br> properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());<br> // 缓冲区大小<br> properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);<br> // 批次大小<br> properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);<br> // linger.ms<br> properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);<br> // 压缩<br> properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");<br> // 1 创建生产者<br> KafkaProducer&lt;String, String&gt; kafkaProducer = new KafkaProducer&lt;&gt;(properties);

<br>

<br>

<br>

开启幂等性

开启事务必须开启幂等性

必须自定义一个唯一的transactional.id

如果未启动幂等性：max.in.fight.requests.per.connection设置为1 如果启动幂等性：max.in.fight.requests.per.connection需要小于等于5

<br>

1. 首先启动的时候将broker启动信息在zk里面注册 2. 然后抢占，看谁抢的快，谁就是监控的controller 3. 监控的controller监控brokers的节点变化，然后选举leader 选举规则是：按照isr中存活前提，然后按照在ar中的顺序 （isr是能互相连通的所有节点，ar是kafka中所有的副本，or是follower和leader副本同步延迟过多的副本 ar=isr+or） 排在ar前面的优先。 leader负责处理读写操作，而follower只负责副本数据的同步 4. 选举完，controller会将选举的信息上传到zk中 5. 其他controller会从zk同步信息 6. 如果有一个节点挂了，监听的controller监听到了变化会去获取isr，然后重新选举新的leader 再更新leader和isr

<br>

follower故障处理细节

leader故障处理细节

手动调整分区副本，创建副本存储计划

leader Partition自动平衡

增加副本因子

首先kafka中topic是逻辑上的概念，partition是物理上的概念，每个partition对应一个log文件，数据是不断追加到log文件的末尾的。为了防止文件过大定位太慢，kafka使用分片和稀疏索引机制，每个log分为多个segment，segment中包括index文件，log文件和timeindex文件，文件命名规则是：topic名称+分区序号

log文件和index文件详解

日志保存时间默认为7天，一旦超过了设置的时间有两种清理策略 1. delete 2. compact

delete是将过期数据删除

compact是将相同key的不同value值只保留最后一个版本

1. kafka本身是分布式的集群，使用分区技术，并行度高

2. 读数据使用稀疏索引，快速定位

3. 顺序写磁盘，kafka写文件一直是追加写，而不是更新

4. 页缓存和零拷贝技术 kafka生产者的数据会直接写入到linux系统内核的页缓存上，页缓存上数据什么时候落盘到file文件由linux系统确定，然后kafka消费者消费数据的时候先去页缓存上拿取数据，再去file的segment中读取数据，然后直接走网卡到消费者，不做任何处理。因为kafka拿取数据的时候不进行处理，就不走应用层，而是直接将数据网卡给到消费者

1. 生产者向每一个分区的leader发送数据，每一个分区的follower去拉取自己分区的leader的数据作为副本 2. 消费者消费数据，每个分区的数据只能由一个消费者组中的一个消费者消费 每个消费者的offset存放在kafka的_consumer_offsets主题下

每个消费者负责消费不同分区的数据，一个分组只能由一个组内的消费者消费

<br>

<br>

1. 首先选择coordinator节点，用这个消费者组的groupid%50获取哪号分区作为这个消费者组的老大

2. coordinator把要消费的topic情况发送给消费者组中的leader消费者，这个leader消费者是随机选出来的

3. 消费者leader会指定消费方案

4. leader将消费方案发送给coordinaor，然后coordinator将消费方案下发给哦各个consumer

5. 注意注意注意！！！！ 这里的每个消费者都会跟coordinator保持心跳，一旦超过45s这个消费者就会被移除，会触发再平衡 这里如果消费者处理数据的时间过长大于5分组，也会触发再平衡

<br>

首先消费者发送消费请求给NetWork客户端，发送请求导对应分区，请求返回一个回调函数OnSuccess，放到一个completeFetches队列中，队列中会有很多返回的数据，然后消费者向这个队列中拉取数据，默认500条拉取一次。 这里注意：每批次抓取的大小默认1字节，也就是当分区内有一个字节的数据就直接拉取，数据最小值超时时间默认是500ms，也就是500ms没有拉取导小于设置的抓取大小的数据就直接都拉取到对列。每批次拉取的最大大小是50m

分区按照序号排序，分区数/消费者数获取每个消费者可以消费多少个分区然后分配

再分配： （1）停止掉0号消费者，快速重新发送消息观看结果（45s以内，越快越好）。 1号消费者：消费到3、4号分区数据。 2号消费者：消费到5、6号分区数据。 0号消费者的任务会整体被分配到1号消费者或者2号消费者。 说明：0号消费者挂掉后，消费者组需要按照超时时间45s来判断它是否退出，所以需要等待，时间到了45s后，判断它真的退出就会把任务分配给其他broker执行。 （2）再次重新发送消息观看结果（45s以后）。 1号消费者：消费到0、1、2、3号分区数据。 2号消费者：消费到4、5、6号分区数据。 说明：消费者0已经被踢出消费者组，所以重新按照range方式分配。

轮询分配

再分配: （1）停止掉0号消费者，快速重新发送消息观看结果（45s以内，越快越好）。 1号消费者：消费到2、5号分区数据 2号消费者：消费到4、1号分区数据 0号消费者的任务会按照RoundRobin的方式，把数据轮询分成0 、6和3号分区数据，分别由1号消费者或者2号消费者消费。 说明：0号消费者挂掉后，消费者组需要按照超时时间45s来判断它是否退出，所以需要等待，时间到了45s后，判断它真的退出就会把任务分配给其他broker执行。 （2）再次重新发送消息观看结果（45s以后）。 1号消费者：消费到0、2、4、6号分区数据 2号消费者：消费到1、3、5号分区数据 说明：消费者0已经被踢出消费者组，所以重新按照RoundRobin方式分配。

粘性分区，尽量均衡的防止分区

再分配： （1）停止掉0号消费者，快速重新发送消息观看结果（45s以内，越快越好）。 1号消费者：消费到2、5、3号分区数据。 2号消费者：消费到4、6号分区数据。 0号消费者的任务会按照粘性规则，尽可能均衡的随机分成0和1号分区数据，分别由1号消费者或者2号消费者消费。 说明：0号消费者挂掉后，消费者组需要按照超时时间45s来判断它是否退出，所以需要等待，时间到了45s后，判断它真的退出就会把任务分配给其他broker执行。 （2）再次重新发送消息观看结果（45s以后）。 1号消费者：消费到2、3、5号分区数据。 2号消费者：消费到0、1、4、6号分区数据。 说明：消费者0已经被踢出消费者组，所以重新按照粘性方式分配。

维护在_consumer_offsets主题里面，kv方式存储，k是groupid+topic+分区号，vaule就是当前的offset值。每隔一段时间就会对这个数进行compact，保存最新数据

enable.auto.commit 默认值为true，消费者会自动周期性地向服务器提交偏移量。 auto.commit.interval.ms 如果设置了 enable.auto.commit 的值为true， 则该值定义了消费者偏移量向Kafka提交的频率，默认5s。

同步提交offset

异步提交offset

同步提交offset和异步提交offset的相同点是都会选择一批数据中最高的偏移量提交，不同是同步提交会阻塞线程，一直到提交成功，自动失败重试。异步提交没有失败重试可能会提交失败 同步提交必须等待offset提交完再去消费下一批数据 异步提交是发送完提交offset请求就去消费下一批数据

public static void main(String[] args) { // 0 配置信息 Properties properties = new Properties(); // 连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,&quot;hadoop102:9092,hadoop103:9092&quot;); // 反序列化 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,&quot;test3&quot;); // 1 创建消费者 KafkaConsumer&lt;String, String&gt; kafkaConsumer = new KafkaConsumer&lt;&gt;(properties); // 2 订阅主题 ArrayList&lt;String&gt; topics = new ArrayList&lt;&gt;(); topics.add(&quot;first&quot;); kafkaConsumer.subscribe(topics); // 指定位置进行消费 Set&lt;TopicPartition&gt; assignment = kafkaConsumer.assignment(); // 保证分区分配方案已经制定完毕 while (assignment.size() == 0){//如果大小等于0，就说明没有获取任何的分区分配方案 kafkaConsumer.poll(Duration.ofSeconds(1)); assignment = kafkaConsumer.assignment(); } // 指定消费的offset for (TopicPartition topicPartition : assignment) { kafkaConsumer.seek(topicPartition,600); } // 3 消费数据 while (true){ ConsumerRecords&lt;String, String&gt; consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord&lt;String, String&gt; consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } }

public static void main(String[] args) { // 0 配置信息 Properties properties = new Properties(); // 连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;hadoop102:9092&quot;); // key value反序列化 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, &quot;test2&quot;); // 1 创建一个消费者 KafkaConsumer&lt;String, String&gt; kafkaConsumer = new KafkaConsumer&lt;&gt;(properties); // 2 订阅一个主题 ArrayList&lt;String&gt; topics = new ArrayList&lt;&gt;(); topics.add(&quot;first&quot;); kafkaConsumer.subscribe(topics); Set&lt;TopicPartition&gt; assignment = new HashSet&lt;&gt;(); while (assignment.size() == 0) { kafkaConsumer.poll(Duration.ofSeconds(1)); // 获取消费者分区分配信息（有了分区分配信息才能开始消费） assignment = kafkaConsumer.assignment(); } HashMap&lt;TopicPartition, Long&gt; timestampToSearch = new HashMap&lt;&gt;(); // 封装集合存储，每个分区对应一天前的数据 for (TopicPartition topicPartition : assignment) { timestampToSearch.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000); } // 获取从1天前开始消费的每个分区的offset Map&lt;TopicPartition, OffsetAndTimestamp&gt; offsets = kafkaConsumer.offsetsForTimes(timestampToSearch); // 遍历每个分区，对每个分区设置消费时间。 for (TopicPartition topicPartition : assignment) { OffsetAndTimestamp offsetAndTimestamp = offsets.get(topicPartition); // 根据时间指定开始消费的位置 if (offsetAndTimestamp != null){ kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset()); } } // 3 消费该主题数据 while (true) { ConsumerRecords&lt;String, String&gt; consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord&lt;String, String&gt; consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } }

重复消费

漏消费

解决漏消费和重复消费