Kafka总结文档

一、概述

// 主要见讲义第1-4页

主要用来缓存数据。主要被当作消息队列

1.1、应用场景

• 缓冲/消峰：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
• 解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。【就是可以暂存数据在消息队列中，消费者需要什么数据就取什么，不需要点对点的建立多条生产者-消费者通讯】
• 异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后在需要的时候再去处理它们。

1.2、消息队列的两种模式

点对点模式会删除数据，发布订阅模式不会删除消息队列中数据

1.3、基础架构

• 分区
• 消费者组
• 副本
• 用到zk。也可以不用zk，就是kraft模式
• 生产者和消费者只针对leader操作【leader是主题在存储时候，副本才有的概念】

二、部署和配置

不需要集群中所有机器都配置kafka，但是配置的kafka就是用作上面图中的broker，用来存储主题数据

另外kafka集群中的机器也可以用命令行操作命令充当生产者和消费者

我们此次，在三台机器上都安装了kafka

需要注意的是！在启动kafka之前一定要先启动zookeeper，如果关闭也是先关闭kafka，并且等他一会再关闭zookeeper

//主要看讲义的第5-7页

1. 先解压tar.gz (tgz后缀) 安装包
2. 进入config修改配置文件。主要修改server.properties - 修改broker.id 每台机器都要不一样！！！！！！注意- 修改log.dir，默认是inux下的tmp临时路径，会被定期删除，我们更改成自己的路径- 修改zookeeper.connect，需要用自己的主机地址，默认端口是2181，zookeeper提供客户端访问的端口。我们需要在最后再加上/kafka，因为zookeeper是目录树结构，如果不加这个文件夹，他会把收到的信息打散到zookeeper中
3. 分发安装包。修改每台机器中server.properties的broker.id
4. 配置环境变量
5. source一下环境变量
6. 先启动zookeeper，再启动kafka集群

三、Kafka命令行操作

这一部分就是讲，怎么用命令行创建主题，查看主题，等一些关于主题的操作 ：bin/kafka-topics.sh

或者启动生产者服务，将数据写进主题中 ： bin/kafka-console-producer.sh

以及消费者怎么查看主题中的数据 ： bin/kafka-console-consumer.sh

但是上面三个都需要声明连接的kafka broker主机名，和端口号。：

-- bootstrap-server hadoop102:9092//这里就是举了一个例子，连接的是kafka集群中hadoop102的broker

3.1、topic命令行操作

// 见讲义的第8-9页

3.2、生产者命令行操作

// 见讲义的第9页

3.3、消费者命令行操作

// 见讲义的第10页

四、Kafka生产者

4.1、生产者消息发送流程

// 主要见讲义的第10页

主要涉及到两个线程：main线程和sender线程

下面章节会仔细讲这里涉及到的重点：

• main线程的send(produceRecord)代码的编写
• 和集群建立联系
• 应答acks参数
• 以及重要参数batch.size和linger.ms
• 需要注意的是！！！一个分区创建一个队列！！！
• sender线程中有一个请求队列，每一个请求队列可以缓冲最多5个请求，不需要等前面的得到应答再发第二个请求

同时需要明确一个事情，就是这章节讲的东西都不是发生在kafka集群，也就是broker上面的

都是生产者这一块需要实现或者配置的

（1）异步发送API编写

// 主要看讲义的第12-13页

异步什么含义，可以去看讲义，有手写笔记。其实就是可以不用等上一次写的数据都被kafka从内存，也就是双端队列中拿走，就可以继续写数据到双端队列。

1. 导入kafaka依赖
2. 编写代码 - 创建kafka生产者的配置对象properties，主要用来存放参数，然后将其放进kafkaproducer对象中- 给properties添加配置信息，必须配置的是bootstrap-server连接的主机，以及key和value的序列化。因为生产者对象其实就是一堆k-v值- 调用send方法，发送数据
3. 调用send方法，发送数据
4. 最后关闭资源
5. 直接运行程序即可，如果消费者打开着，就可以看到生产者写进去的数据

（2）带回调函数的异步发送API编写

// 主要看讲义的第13-15页

主要就是数据发送结束后，返回数据的相关信息到控制台

在send 方法中加上一个callback方法，里编写自己想要的内容

// 添加回调
 kafkaProducer.send(newProducerRecord<>("first","atguigu "+ i),newCallback(){// 该方法在 Producer 收到 ack 时调用，为异步调用@OverridepublicvoidonCompletion(RecordMetadata metadata,Exception exception){if(exception ==null){// 没有异常,输出信息到控制台System.out.println(" 主题： "+ metadata.topic()+"->"+"分区："+ metadata.partition());}else{// 出现异常打印
            exception.printStackTrace();}}});

（3）同步发送API编写

// 主要看讲义的第15-13页

就是在异步发送的基础上，再调用一下get()方法

kafkaProducer.send(newProducerRecord<>("first","kafka"+ i)).get();

4.2、生产者分区

这个指的是将要存放的数据，分区存放到不同的broker中【前提是那个topic在创建的时候是分好区的】

1、分区策略

有三种分区策略：

1. 分别是指明自己将要存放的数据存放到哪个分区。
2. 或者没有指明分区的值，但是提供了key。
3. 最后一个就是什么也没制定，但是会尽可能都划分到一个区

所以说前两种都需要提供key值，我们4.1节的代码中默认使用的send中ProducerRecord只是提交value值

//主要见讲义第18-21页

其实在实际编写代码的时候，就是改send部分，他底层源码会按照上面的三个策略去帮你分配进入哪个分区

2、自定义分区

//主要见讲义第21-23页

主要是：

1、定义类实现 Partitioner 接口。

2、重写 partition()方法。

代码中实际完成的就是重写partition方法中的几个步骤:

• 获取消息，将参数的value值变成tostring
• 然后根据需求，去分析value，然后返回不同情况返回不同的partition值
• 最后将这个类的全类名，放在生产者的代码中，添加一个参数// 添加自定义分区器properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.atguigu.kafka.producer.MyPartitioner");

4.3、生产经验

1、提高吞吐量

//主要见讲义第24-25页

在生产者代码中可以调整参数：

• 批次大小
• 等待时间
• 缓冲区大小
• 压缩

2、数据可靠性

//主要见讲义第25-28页

这里主要涉及到的就是应答机制。不同的应答适合不同的场景，而应答在kafka这里为了避免数据重复有引入两个概念：幂等性和事务

一共有三个应答等级0，1，-1

但是各有利弊，其中较为完善的“ -1，也就是all ” 中，还是会有一个问题。

Leader收到数据，所有Follower都开始同步数据， 但有一个Follower，因为某种故障，迟迟不能与Leader进行 同步，那这个问题怎么解决呢？

提出了ISR。基于此：

数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

但是还会有一个数据可能重复的问题。

其实默认就是-1的机制，但是我们可以在生产者代码中进行配置
    
// 设置 acks
 properties.put(ProducerConfig.ACKS_CONFIG,"all");

3、数据去重

//主要见讲义第29-31页

这里主要是幂等性和事务

默认幂等性是开的。

需要明确，开启事务，必须开启幂等性。

其实事务很好理解，就是相当于对每一次的请求，还有持久化，也就是数据在kafka中的docker落盘情况记录一次

这样就避免了如果这个docker出问题崩掉了，也就是会话断开了，幂等性就会认为这之后再来点的数据就是不重复的了，因为PID不同了

在生产者代码中，实际就是引入关于事务的一个API，在讲义中可以看看

一定要设置一个事务id！！！！要不一样的

这里举一个例子：

// 设置事务 id（必须），事务 id 任意起名
properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction_id_0");// 3. 创建 kafka 生产者对象KafkaProducer<String,String> kafkaProducer =newKafkaProducer<String,String>(properties);// 初始化事务
kafkaProducer.initTransactions();// 开启事务
kafkaProducer.beginTransaction();try{// 4. 调用 send 方法,发送消息for(int i =0; i <5; i++){// 发送消息
         kafkaProducer.send(newProducerRecord<>("first","atguigu "+ i));}// 提交事务
     kafkaProducer.commitTransaction();}catch(Exception e){// 终止事务
     kafkaProducer.abortTransaction();}finally{// 5. 关闭资源
     kafkaProducer.close();}}

4、数据有序 & 数据乱序

其中数据乱序是面试重点！！！

//主要见讲义第32页

分为单分区有序和多分区有序。但是多分区有序这个需要等所有分区数据都到齐才能重新排序，因为分区间是无序的

而单分区有序是有条件的，没有开启幂等性的时候，那就是一个请求处理完才会来一个请求，这样的话两个请求之间的顺序不会乱，因为如果第一个请求没有完成我们不管的话，直接开始接受第二个请求去传数据，那第一个请求的数据会在之后发送，那么就会造成这个分区的数据混乱无序掉

但是开了幂等性后就没事了，比如第三个请求挂掉，但是他会记录顺序，之后先接受第四个和后面的请求，之后接受了第三个请求后，在broker中重新排序

五、Kafka Broker

broker这里主要明白他的存储原理，和zookeeper是怎么交互的，怎么用到zookeeper的

另外就是明确里面很重要的一个东西，每一个主题会进行分区，每个分区会有很多个副本，那这些副本中怎么选举leader，生产者和消费者只是针对leader进行拉取和写入数据

副本中leader和follower故障怎么处理，怎么负载均衡，怎么分配副本

另外就是面试重点！高效率的读写是怎么达到的，在讲义的4.5节

5.1、Zookeeper存储Kafka信息

这块查看zookeeper中节点存储的信息，可以用一个app，叫prettyZoo，可以看到内部信息

主要明白zookeeper中存储的kafka信息都有什么，和什么含义

// 主要见讲义的第33页

这里重点理解一下controller
因为下面broker的工作流程也是很多用到这个东西
要明白最后kafka集群，也就是broker集群中只会有一个节点上的controller模块会被选中当作整个集群的controller
其中，每个节点都有一个controller模块，注意是模块，他是实际可以决定哪个副本是leader的
但是由于每一个节点都有这么一个模块
所以就会抢着注册controller，谁抢到谁就说的算

5.2、broker工作流程

// 主要见讲义第33-35页

1. broker启动 后在zk中注册
2. zookeeper中的 ids 会记录正常的服务器
3. 每台broker抢着注册controller，controller谁先注册，谁说了算，假如broker0抢到了【每一台broker都有一个controller】
4. 那broker0上的controller模块监听brokers节点变化【监听各个节点的状态，从zoo中ids可以监听到】
5. Controller决定Leader选举，决定出来副本中谁是leader，这里说的副本是每一个主题每一个分区都会有几个副本，这几个副本中就有一个leader，剩下的就是follower
6. Controller将节点信息上传到ZK
7. 其他contorller从zk同 步相关信息【第5和6其实就是一个同步信息的过程，防止之后有节点挂点，信息不全】
8. 假如broker1这个leader挂了，controller监听到后
9. 从zookeeper中获取ISR，和AR
10. 选举新的Leader（在 isr中存活为前提，按照 AR中排在前面的优先）
11. 更新Leader及ISR

5.3、节点服役和退役

这个其实主要就是生产环境中，可能有时候突然需要新增加broker节点，或者退役节点

这就需要操作一下

// 主要看讲义的第35-38页

1、服役

• 准备一个新的服务器，下载好jdk，kafka等环境。【实际生产中确实需要一个干净的服务器，但是平常学习直接克隆】
• 打开hadoop105这个克隆好的，修改静态IP，主机名【之前主机映射文件也改好了hadoop105和ip的了】
• 重新启动
• 重点修改克隆后机器上的配置文件，和其余文件1. 修改 haodoop105 中 kafka 的 broker.id 为 3。2. 删除 hadoop105 中 kafka 下的 datas 和 logs。
• 最后单独启动hadoop105的kafka就相当于这个节点上线了

但是之前存储的数据只是存放在hadoop102-104上，怎么进行改变？

执行负载均衡操作

1. 创建一个要均衡的主题：一个json文件{"topics":[{"topic":"first"}//可以有多个主题],"version":1}
2. 命令行用上面的json文件生成一个负载均衡的计划。 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
3. 执行上面生成的计划- 先复制上面生成的计划到一个新的json文件- 然后执行命令行命令，执行计划bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

2、退役

退役其实也是需要重新用到上面的负载均衡计划

把要退役节点数据均衡走

然后再退出节点的kafka服务即可

5.4、副本

// 主要看讲义的第38-44页

1、首先明确副本是针对谁的：

​ 每一个主题的每一个分区会有几个副本，不同副本存放在不同的broker中。所以分区可以有很多个，但是副本个数最好不要超过broker个数。并且从这几个副本里面选举一个leader

2、Kafka分区中的所有副本统称AR

AR=ISR+OSRISR，表示和 Leader 保持同步的 Follower 集合。
// 如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。OSR，表示 Follower 与 Leader 副本同步时，延迟过多的副本

1）leader选举流程

//其实在broker中的工作流程中有讲到，不过讲义的第39-40举了一个例子，可以看看详细过程

2）Leader和Follower故障处理细节

//看讲义的第39-40页

这里面有两个标记值，LEO和HW

follower故障后再恢复的时候：是Follower会读取本地磁盘记录的 上次的HW，并将log文件高于HW的部分截取掉，从HW开始 向Leader进行同步。等该Follower的LEO大于等于该Partition的HW，即 Follower追上Leader之后，就可以重新加入ISR了。

leader故障后：其余的Follower会先 将各自的log文件高于HW的部分截掉，然后从新的Leader同步数据。【追求的数据一致性，不能保证数据不会丢失或者重复】

3）分区副本分配

//讲义的第41-42页

假如有四个broker，分区数大于服务器台数的时候。就会出现每一个broker出现多个分区

那这种情况如何分配副本呢？是有一定规律的

但是也可以手动调整分区副本，因为有的时候新服役的机器性能比较好
可以多分配一些
这个其实和节点服役退役是一样的流程，去创建计划然后执行计划

//讲义的第43页

4）Leader partition负载均衡

//讲义的第44页

如果某些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，

正常来说，一般AR的顺序，也就是里面副本存放的broker主机id是不会更换顺序的

所以AR中的第一个id，比如1，就是该分区对应副本中，leader的那个副本所在的broker节点id，，但是如果实际的leader不是在这个上面，比如是0，这个不是1，那就是1发生过宕机，所以1的不平衡数加一

5.5、文件存储

一个partition对应一个log，每一个log是由很多个segment组成，而segment又由index，log，timeindex三个文件组成

文件存储这里，需要明白是怎么利用到index文件中的offset，找到log文件中对应的数据

并且记住！！index是稀疏索引：就是每往log中写入4kb数据，才会往index中写入一条索引

//讲义的第45-46页

5.6、文件清理策略

//讲义的第47-48页Kafka 中默认的日志保存时间为 7 天
删除日志，包括基于时间和基于大小两种

日志压缩，这个压缩和平常说到的压缩不一样！！！！！！！只留下最新的，类似是覆盖

5.7、高效读写数据【面试重点】

//讲义的第48-49页

1. Kafka 本身是分布式集群，可以采用分区技术，并行度高
2. 读数据采用稀疏索引，可以快速定位要消费的数据
3. 顺序写磁盘【顺序比随机写快很多】
4. 页缓存 + 零拷贝技术

六、Kafka消费者

消费者组中的每一个消费者只能消费一个主题的分区

哪怕只有一个消费者，那这个消费者也是一个消费者组，在写api代码的时候需要设置一个id

1、消费方式

拉pull 和 推push两种模式

主要由消费者决定自己的需求，因为broker没法决定每个消费者不同的速度需求，所以采取拉取数据的方式

但是也有不足之处，就是如果broker中没有数据，消费者会陷入循环中，一直返回空数据

2、消费者工作流程

1. 消费者组初始化，主要涉及每个消费者消费哪个分区的分配，还有一些参数的配置
2. 详细消费流程

// 主要见讲义的第50-52页

这里有两个重点！！！！！！！！！！
// 在初始化过程中，需要知道coordinator节点如何进行选择的// 还有触发再平衡的两个条件，就是两个时间参数，一个3s，一个45s，一个5min

1）消费者组初始化【面试重点】

这里主要是知道消费者组中是怎么控制每一个消费者是去消费哪个分区数据的

每一个broker节点都有一个coordinator，这个可以辅助消费者组选择出来怎么消费

然后类似事务中有一个特殊的主题，默认50个分区，然后每一个消费者组有一个独立的id。由id的hash值去%50

得到的值就是选中负责的coordinator，然后coordinator会随机选择消费者组中的一个消费者是leader，然后把这个leader指定的方案，发送给消费者组中的所有消费者。也会接受提交的offset

具体流程如下图：

2）消费流程

• 发送拉取数据请求
• 回调信息，将数据写入一个消息队列
• 然后对队列中的数据反序列化，经过拦截器处理数据
• 消费者组拉取数据

3、消费者API【代码实现】

//主要见讲义第53-59页

如果只是订阅主题，就是找到想要消费的主题，把主题中来的数据都消费【获取】，那在注册要消费主题时候用subscribe()
如果是订阅分区，就是消费主题中具体的分区，用assign()

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

1、独立消费者案例【订阅主题】

1. 创建消费者配置对象properties
2. 给消费者配置对象添加参数- 连接参数- 配置反序列化- 一定不要忘了！！！配置消费者组id properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
3. 创建消费者对象//1 创建一个消费者KafkaConsumer<String,String> kafkaConsumer =newKafkaConsumer<>(properties);
4. 注册要消费的主题，就是subscribe//2 订阅主题ArrayList<String> topics =newArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);
5. 拉取数据，消费数据，就是poll//3 消费数据while(true){// 设置 1s 中消费一批数据ConsumerRecords<String,String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));// 打印消费到的数据for(ConsumerRecord<String,String> consumerRecord : consumerRecords){System.out.println(consumerRecord);}}

2、独立消费者案例【订阅分区】

就是在订阅部分是不一样的

不是subscribe，而是assign

// 消费某个主题的某个分区数据ArrayList<TopicPartition> topicPartitions =newArrayList<>();
topicPartitions.add(newTopicPartition("first",0));//说明要订阅的主题，和其中的分区
kafkaConsumer.assign(topicPartitions);//对比一下单纯的订阅主题ArrayList<String> topics =newArrayList<>();
topics.add("first");
kafkaConsumer.subscribe(topics);

3、消费者组案例

其实就是定义很多个消费者代码

主要他们的group的id一样就行

他们就会被分配到同一个消费者组

4、分区的分配以及再平衡

//主要见讲义第59-65页Kafka有四种主流的分区分配策略： Range、RoundRobin、Sticky、CooperativeSticky。
    
这里提到的再平衡就是，按照45s这个标准判断这个broker是否退出，如果超过45s这个消费者还是没有进行拉取数据进行消费
那么，这个消费者本应该消费的数据，就会被重新分配到其他机器上，这些数据分配到其他机器上的方案，和整体的方案是一致的
    
上面那种情况是针对45s之前还是存活工作的，如果被检查到上次就已经退出不工作了，那么这个broker就不会被分配数据，就相当于集群中没有这个机器，不会出现上面那段描述的情况
    
//这个再分配，在下面的集中分配策略中，讲义和自己的手写笔记也有详细描写，可以好好看看

可以通过配置参数partition.assignment.strategy，修改分区的分配策略。默认策略是Range + CooperativeSticky。
Kafka可以同时使用多个分区分配策略。

分区方案是怎么发送的呢？

回忆一下消费者的工作流程，每一台broker都有一个coordinator，根据消费者组的id，去选出来是哪个broker的coordinator用来做决定。coordinator会随机选择一个消费者，让其当作leader，之后这个leader的消费者，就负责制定消费方案，将方案发给coordinator，coordinator会把方案发给消费者组中的每一个消费者

那么这里面的leader在指定消费方案的时候，是怎么指定分区的

1、Range

//见讲义的第60-62页

• 先对同一个topic里的分区进行排序，消费者按字母也进行排序
• 通过 partitions数/consumer数 来决定每个消费者应该 消费几个分区。如果除不尽，那么前面几个消费者将会多 消费 1 个分区。【这里说的多消费一个分区，是在最开始多消费，比如就是第一个消费者消费0，1，2；之后第二个消费者消费3，4】

这样的分区策略如果topic越多，那么多消费的那个消费者就会明显多消费。

容易造成数据倾斜

这种策略是并不需要指定的，默认就是。只要生产者对每个分区都发过一些数据，就可以检验到

2、RoundRobin

//见讲义的第62-63页

• 这个是针对集群中所有的topic而言【！！！注意！！！这里是针对所有topic中的partition一起排序了！！！】
• 把所有的 partition 和所有的 consumer 都列出来，然后按照 hashcode 进行排序，最后 通过轮询算法来分配 partition 给到各个消费者【这里说的轮询，就是从上到下一个个把partition分配给消费者，可以具体看讲义中的图】

这个就需要更改一下分配策略了。就是在代码中更改就好。最好多看看官网，因为每个版本的kafka参数可能不太一样

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

3、 Sticky 以及再平衡

//见讲义的第64-65页

• 也是针对集群中所有的topic而言
• 首先会尽量均衡的放置分区 到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

【这一部分看讲义手写笔记吧。类似range，但是不是有序的，不过之后的版本好像是有规律分布】

5、offset位移

这个offset主要是用来记录消费者消费到哪里了。

在kafka0.9之前是保存在zookeeper中，后来是保存在内置的一个特殊topic中，类似前面生产者的事务id

主题里面采用 key 和 value 的方式存储数据。

key 是 group.id+topic+ 分区号，value 就是当前 offset 的值。

每隔一段时间，kafka 内部会对这个 topic 进行 compact，也就是每个 group.id+topic+分区号就保留最新数据【这里说到的压缩不是平常那个compact压缩，其实就是“覆盖”】

5.1、查看消费offset

// 在讲义的第66页

默认是不能查看消费系统主题数据的，需要改成false: config/consumer.properties 中添加配置 exclude.internal.topics=false，
如果是命令行启动消费者，最好手动加上group id,要不系统自动给你加一个，你不知道加的是啥样的id

5.2、自动提交offset

//讲义的第67-68页

也就是默认打开这个自动提交offset，每隔5秒提交一次

默认是打开的，我们之后写手动提交的时候可以关闭，也可以自己设置自动提交的间隔时间。这个都是在消费者代码中写的

但是需要注意的是，不同版本的参数可能不一样，要多看官网

// 是否自动提交 offset
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);// 提交 offset 的时间周期 1000ms，默认 5s
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);//1000的单位是ms

5.3、手动提交offset

//讲义的第71-73页

之所以提到手动提交offset这个东西，是因为自动提交是基于时间的，我们很难把握offset提交的时机
所以还是有API可以让我们去手动提交的

比如我们可以设置，消费一条数据就提交一次offset，避免了自动提交中可能的提前消费，有时候处理时间过程，但是它已经提交了offset，但是实际它并没有消费到这个提交的offset
    
// 手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。

• commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。

• commitAsync（异步提交） ：发送完提交offset请求后，就开始消费下一批数据了【可能offset没提交成功，但也开始消费下一批数据】

生产中大多是异步提交的，追求的是效率

其实实现的流程十分简单

1. 先是将自动提交offset关闭
2. 然后在消费数据的代码部分设置同步提交或者异步提交的代码

// 是否自动提交 offset
 properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);// 同步提交 offset
 consumer.commitSync();// 异步提交 offset
 consumer.commitAsync();

5.4、指定offset消费

//讲义的第73-75页

这一部分主要是指定一开始从哪里开始消费，就是针对消费者怎么利用offset

（1）earliest：自动将偏移量重置为最早的信息位置，–from-beginning。

（2）latest（默认值）：自动将偏移量重置为最新发送来的信息位置【偏移量】

其实上面说的这个就是，分区中的数据位置，一个是从最头的位置开始，另外一个是当前时刻，注意是时间的当前时间下，生产者可能又发送新数据来，那我们就送这里还是进行往后消费，这里就是此次消费者组offset的起点

代码实现就是seek，然后获取分区分配信息，用seek去从分区中我们指定的offset中开始消费

// 这里还是看讲义的第73页代码吧

5.5、指定时间消费

//看讲义的第74页代码和手写笔记

其实就是想办法找到时间和offset的联系，想要通过时间获取到offset
这一部分比较晦涩，好好看看代码吧

6、漏消费和重复消费

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

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

7、消费者事务

//讲义的第76页

那么需要Kafka消费端将消费过程和提交offset
过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质
比如（MySQL）

8、数据积压【生产中很重要！！！】

//讲义的第76页

可能两点造成：
    1、kafka集群性能所限
    2、拉取数据不及时，拉的数据一次较少或者时间间隔过长。也可以这么说，就是消费的及时，但是后续发给消费者目的地的效率不高，目的地

七、Kafka-Eagle监控【efak】

// 见讲义第77-81页Kafka-Eagle 框架可以监控 Kafka 集群的整体运行情况
Kafka-Eagle 的安装依赖于 MySQL，MySQL 主要用来存储可视化展示的数据。

1. 安装mysql
2. kafka内存从1G调整到2G，并且暴露一个用来监控kafka的端口号，9999
3. 安装kafka-Eagle
4. 修改其中配置文件件/opt/module/efak/conf/system-config.properties - 更改监控集群的信息- 更改保留kafka的offset的选线，删掉zk- 配置mysql的连接参数
5. 添加环境变量
6. 启动bin/ke.sh start【但是需要先开启zookeeper和Kafka】
7. 命令生成后，会提供一个网址，和账号密码。进去后可以看到各种详细信息、可视化界面，也可以直接在界面中多主题进行创建等操作

八、Kafka-kraft模式

这个模式是不需要借助zookeeper

以后实际生产会慢慢转向此种模式。在Kafka3.0之后提供的，但是目前还可以兼容两种模式，用不用zookeeper都可以

看你怎么配置和使用

//见讲义第82-84页

看怎么安装部署的

九、调优【涉及到调优讲义】

主要涉及四部分：

1. 硬件配置怎么配，需要什么样的参数，给什么样的硬件条件
2. 生产者，消费者，broker服务端。这三部分的调优和生产经验，在前面文档都讲过了，都在api部分后面那些就是
3. 总体调优部分，其中数据精确一次这个内容可以好好看一下，容易面试问！！还有合理分配分区数的计算公式，单条日志大小
4. 压测，kafka提供了命令行脚本，可以进行压测

//主要看第一、三、四，这三个部分

十、源码【涉及到源码讲义】

另外看源码讲义

其中生产者和消费者，是用java写的，broker，是在源码中的core文件中，用scala语言写的
5.png" alt=“image-20240101155217055” style=“zoom:50%;” />

七、Kafka-Eagle监控【efak】

// 见讲义第77-81页Kafka-Eagle 框架可以监控 Kafka 集群的整体运行情况
Kafka-Eagle 的安装依赖于 MySQL，MySQL 主要用来存储可视化展示的数据。

1. 安装mysql
2. kafka内存从1G调整到2G，并且暴露一个用来监控kafka的端口号，9999
3. 安装kafka-Eagle
4. 修改其中配置文件件/opt/module/efak/conf/system-config.properties - 更改监控集群的信息- 更改保留kafka的offset的选线，删掉zk- 配置mysql的连接参数
5. 添加环境变量
6. 启动bin/ke.sh start【但是需要先开启zookeeper和Kafka】
7. 命令生成后，会提供一个网址，和账号密码。进去后可以看到各种详细信息、可视化界面，也可以直接在界面中多主题进行创建等操作

八、Kafka-kraft模式

这个模式是不需要借助zookeeper

以后实际生产会慢慢转向此种模式。在Kafka3.0之后提供的，但是目前还可以兼容两种模式，用不用zookeeper都可以

看你怎么配置和使用

//见讲义第82-84页

看怎么安装部署的

九、调优【涉及到调优讲义】

主要涉及四部分：

1. 硬件配置怎么配，需要什么样的参数，给什么样的硬件条件
2. 生产者，消费者，broker服务端。这三部分的调优和生产经验，在前面文档都讲过了，都在api部分后面那些就是
3. 总体调优部分，其中数据精确一次这个内容可以好好看一下，容易面试问！！还有合理分配分区数的计算公式，单条日志大小
4. 压测，kafka提供了命令行脚本，可以进行压测

//主要看第一、三、四，这三个部分

十、源码【涉及到源码讲义】

另外看源码讲义

其中生产者和消费者，是用java写的，broker，是在源码中的core文件中，用scala语言写的