Kafka3.0理论学习

一、kafka概述

1.1、定义

Kafka传统定义：Kafka是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域。

发布/订阅：消息的发布者不会将消息直接发送给特定的订阅者，而是将发布点消息分为不同的类别，订阅者只接收感兴趣的消息。

Kafka最新定义：Kafka是一个开源的分布式事件流平台（Event Streaming Platform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。

1.2、消息队列

常见的消息队列产品有 Kafka、ActiveMQ、RabbitMQ、RocketMQ等。

1.2.1、传统消息队列的应用场景

主要应用场景：缓存/消峰、解耦、异步通信

1.2.2、消息队列的两种模式

1）点对点模式

2）发布/订阅模式

1.3、Kafka基础架构

二、Kafka快速入门

2.1、安装部署

自行参考其他博客

2.2、命令行操作

2.2.1、主题命令行操作

1）查看操作主题命令参数

2）查看当前服务中所有到topic

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list
1

2.2.2、生产者命令行操作

bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first 
1

2.2.3、消费者命令行操作

bin/kafka-console-consumer.sh  --bootstrap-server	hadoop102:9002 --topic first
1

三、Kafka生产者

3.1、生产者消息发送流程

​ 在消息发送的过程中，涉及到了两个线程-main线程和sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。

3.2、异步发送API

3.2.1、普通异步发送

3.2.2、带回调函数带异步发送

3.3、同步发送API

3.4、生产者分区

3.4.1、分区好处

1）便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。

2）提高并行度，生产者可以以分区为单位发送数据，消费者可以以分区为单位进行消费数据

3.4.2、生产者发送信息带分区策略

3.4.3、自定义分区器

1）需求：

例如我们实现一个分区器实现，发送过来的数据中如果包含zhangsan，就发送0号分区，不包含，就发往1号分区

2）实现

package com.das.pi.kafka;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

public class MyPartitioner implements Partitioner {
    @Override
    public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) {
        //获得数据
        String msgValues = o1.toString();
        int partition;
        if(msgValues.contains("zhangsan")){
            partition = 0;
        }else {
            partition = 1;
        }
        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

3.5、生产经验-生产者如何提高吞吐量

1）batch.size：批次大小,默认16k

2）kinger.ms：等待时间，默认0

3）RecordAccumulator：缓冲区大小，默认32M：buffer.memory

4）compression.type：压缩，默认none，可配置值gzip、snappy、lz4、zstd

3.6、生产经验-数据可靠性

acks：

​ 0:生产者发送过来的数据，不需要等数据落盘应答 （数据可靠性分析：丢数）

​ 1:生产者发送过来的数据，Leader收到数据后应答。（数据可靠性分析：丢数）

​ -1/all：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。注意思考：Leader收到数据，所有Follower都开始同步数据，但有一个Follower，因为某种故障，迟迟不能与leader进行同步，那这个问题怎么解决？

Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步的Follower+Leader集合（Leader：0，isr：0，1，2）

如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。这样就不用等长期联系不上或者已经故障的节点。

3.7、生产经验-数据去重

kafka 0.11版本后，引入了一项重大特征：幂等性和事务。

如何使用幂等性，开启参数enable.idempotence默认为true，false关闭

3.9、生产经验-数据乱序

四、Kafka Broker

4.1、Kafka Broker 工作流程

4.2、生产经验-节点服役和退役

4.2.1、服役新节点

1）创建新的节点，启动kafka

2）执行负载均衡操作

a、创建一个均衡主题

vim  topics-to-move.json

{
	"topics":[
		{"topic":"first"}
	],
	"version":1
}
1
2
3
4
5
6
7
8

b、生成一个负载均衡的计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate
1

c、创建副本存储计划（所有副本存储在broker0、broker1、broker2、broker3中）。

vim increase-replication-factor.json
1

d、执行副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
1

e、验证副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
1

4.2.2、退役旧节点

4.3、Kafka副本

4.3.1、Leader选举流程

4.3.2、Leader和Follower故障处理细节

1）Follower挂了

2）Leader挂了

4.3.3、分区副本分配

4.3.4、生产经验-手动调整分区副本存储

4.3.5、生产经验-Leader Partition 自动平衡

⚠️注意：一般情况不建议把自动再平衡设置为true，会浪费大量的性能

4.3.6、生产经验-增加副本因子

4.4、文件存储

4.4.1、文件存储机制

1）Topic数据的存储机制

2）思考：Topic数据到底存储在什么位置？

查看log文件(反序列化)

kafka.tools.DumpLogSegments --files ./00000000000.log 
1

Log文件和Index文件详解

4.4.2、文件清理策略

1）compact日志压缩 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0r2YtfeF-1682068761959)(https://hnuscwy.work/minio/typroa/20230419105400.png)]

4.5、高效读写数据

4）页缓存+零拷贝技术

参考：图解Kafka的零拷贝技术到底有多牛？https://cloud.tencent.com/developer/article/1421266

五、kafka消费者

5.1、kafka消费方式

5.2、kafka消费者工作流程

5.2.1、消费者总体工作流程

5.2.2、消费者组原理

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

消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

5.2.3、消费者组初始化流程

5.2.4、消费者组详细消费流程

5.3、消费者API

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

​ 创建一个独立消费者，消费first主题中数据。⚠️注意：在消费者API代码中必须配置消费者组id。命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id。

//配置
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_DESRIALIZER_CLASS_CONFIG,StringDeserizalizer.class.getName());
//配置消费组id
propertise.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

//创建一个消费者“”，“hello”
KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<>(properties);

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

//消费数据
while(true){
  ConsumerRecords<String,String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
  for(ConsumerRecord<String,String> consumerRecord : consumerRecords){
    System.out.println(consumerRecord);
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

5.3.2、独立消费者案例（订阅分区）

//配置
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_DESRIALIZER_CLASS_CONFIG,StringDeserizalizer.class.getName());
//配置消费组id
propertise.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

//创建一个消费者“”，“hello”
KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<>(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);
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

5.3.3、消费者组案例

​ 测试同一个主题的分区数据，只能由一个消费者组中的一个消费者进行消费。⚠️注意：创建三个消费者类、且消费者组的id都相同。

5.4、生产经验-分区的分配以及再平衡

参考：kafka 消费者–重平衡策略 https://www.modb.pro/db/126099

5.4.1、Range策略

​ 按照topic维度进行平均分配，将每个topic的partition按照顺序编排好，然后按照消费者数量开始平均分配。假如有一个topic，有m个分区，n个消费者。那么每个消费者至少消费m/n，前面的m%n的消费者多消费一个。

5.4.2、RoundRobin策略

​ 按照分区的维度进行平均分配，将所有的topic的分区数加起来，然后按照消费者数量平均分配。假如某个消费者组，共监听m个分区，n个消费者。那么每个消费者至少消费m/n，前面的m%n的消费者多消费一个。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Ay1E1W0X-1682068761961)(https://hnuscwy.work/minio/typroa/20230420161634.png)]

5.4.3、Sticky策略

​ 粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。

​ 粘性分区是kafka从0.11.x版本开始引入这种分配策略，首先会尽量均衡的放置分区在消费者上面，在出现同一消费组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

⚠️注意：

1）分区尽可能和上一次分配结果保持一致

2）分配尽可能均匀，每个消费者消费的分区数最多相差1

5.4.4、CooperativeSticky策略

​ 这个的分配逻辑是和Sticky一致的，和前面分配逻辑不同的是，前面的都是EAGER协议，CooperativeSticky有两个协议，一个是EAGER，一个是COOPERATIVE。EAGER协议和上面的Sticky分配规则一样，先会放弃所有的分区，等待协调者返回重新分配的分区结果。COOPERATIVE协议就不一样了，它是一个渐进重平衡过程，这个过程可以允许消费者继续保留当前的分区不变化，然后等待协调者重新分配增量的分区。

我们通过一个例子说明一下COOPERATIVE协议下的分配方案：

Eg:两个消费者c1和c2，一个topic三个分区p1,p2,p3。刚开始分配结果是c1[p1,p2],c2[p3]。此时增加了一个消费者c3，按照Sticky分配的结果是：

c1[p1],c2[p3],c3[p2]。不过，在这个过程中，会存在两次重平衡的过程，c1,c2上报分区，获得的分配结果是c1[p1],c2[p3]，这个时候p2没有人消费。第二次重平衡的过程是将p2分配给p3。

5.5、offset位移

5.5.1、offset的默认维护位置

5.5.2、自动提交offset

5.5.3、手动提交offset

//设置手动提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

//同步手动提交
kafkaConsumer.commitSync();

//异步手动提交
kafkaConsumer.commitAsync();
1
2
3
4
5
6
7
8

5.5.4、指定offset消费

//指定位置进行消费
Set<TopicPartition> assignment = kafkaConsumer.asssignment();
//指定消费的offset
for(TopicPartition topicPartition : assignment){
  	kafkaConsumer.seek(topicPartition,100);
}
//保证分区分配方案已经制定完毕
while(assignment.size() == 0){
  kafkaConsumer.poll(Duration.ofSeconds(1));
  assignment = kafkaConsumer.assignment();
}
1
2
3
4
5
6
7
8
9
10
11

5.5.5、指定时间消费

//指定位置进行消费
Set<TopicPartition> assignment = kafkaConsumer.asssignment();

//希望把时间转换为对应的offset
HashMap<TopicPartition,Long> topicPartitionLongHashMap = new HashMap<>();

//封装对应集合
for(TopicPartition topicPartition : assignment){
  topicPartitionLongHashMap.put(topicPartition,System.currentTimeMillis() - 1* 24* 3600 * 1000);
}
  Map<TopicPartition,OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);

//指定消费的offset
for(TopicPartition topicPartition : assignment){
    OffestAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);
  	kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
}
//保证分区分配方案已经制定完毕
while(assignment.size() == 0){
  kafkaConsumer.poll(Duration.ofSeconds(1));
  assignment = kafkaConsumer.assignment();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

5.5.6、漏消费和重复消费分析

重复消费：已经消费了数据，但是offset没提交

漏消费：先提交offset后消费，有可能造成数据的漏消费。

5.6、生产经验-消费者事务

5.7、生产经验-数据积压

六、Kafka-Eagle（EFAK）监控

七、kafka-kraft模式

八、集成SpringBoot

参考：https://juejin.cn/post/7028149679976251422