kafka 安装使用 /springboot整合kafka /消息投递机制以及存储策略 /副本处理机制

一、背景

1、基本信息

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。
百度百科：https://baike.baidu.com/item/Kafka/17930165

2、作用

可以实现支撑高并发、异步解耦、流量削峰、降低耦合度。

1.异步发送短信
2.异步发送新人优惠券
3.处理一些比较耗时的操作
4. 秒杀异步处理，后续给结果通知
5. 火车票

3、kafka性能高体现

消息中间件基于队列模型实现异步/同步传输数据

1.顺序读写； (写数据在 log文件中追加内容)
2.零拷贝；( 由 linux 内核支持)
3.分区存放 ; (集群扩展)
4.分批发送；(高并发时缓冲区缓存投递消息)
5.消息压缩，

4、零拷贝

零拷贝技术： dma + mmap内存映射 + sendfile 实现，仅 linux 支持

实现流程

1.使用到mmap内存映射 用户与内核空间实现共享虚拟内存，不需要在使用cpu将
内核空间的数据拷贝到用户空间。
2.使用sendfile 2.4 版本linux 内核 使用dma技术将硬盘的数据拷备到内核态，在使用dma技术将内核态数据拷贝到网卡。
3.最终只需要做两次上下文切换两次dma数据拷贝（直接内存）不需要cpu拷贝数据
从而减少上下文切换提高cpu的利用率。

名词解释

dma：直接内存拷贝，不需要cpu 调度资源
mmap内存映射：内核空间数据共享给用户空间，不在需要把内核空间数据拷贝到用户空间
sendfile : 可以把数据通过 dma 技术直接拷贝给网卡返回给消费者，不走cpu调度

5、设计原理

1.高吞吐、低延迟：kakfa 最大的特点就是收发消息非常快，kafka 每秒可以处理几十万条消息，它的最低延迟只有几毫秒；
2.高伸缩性：每个主题(topic) 包含多个分区(partition)，主题中的分区可以分布在不同的主机(broker)中；
3.持久性、可靠性：Kafka 能够允许数据的持久化存储，消息被持久化到磁盘，并支持数据备份防止数据丢失，Kafka 底层的数据存储是基于 Zookeeper 存储的，Zookeeper 我们知道它的数据能够持久存储；
4.容错性：允许集群中的节点失败，某个节点宕机，Kafka 集群能够正常工作；
5.高并发：支持数千个客户端同时读写。

6、Kafka消息队列模型

Kafka 的消息队列一般分为两种模式：点对点模式和发布订阅模式

点对点
一个生产者生产的消息由一个消费者进行消费

发布订阅模式
多个生产者产生的消息能够被多个消费者同时消费
使用 topic 主题管理消息

二、安装

由于Kafka是用Scala语言开发的，运行在JVM上，因此在安装Kafka之前需要先安装JDK。

链接：https://pan.baidu.com/s/1ZTMhFIf6iAVDyR6IhPG0Ig
提取码：pg42

2.1、win 安装

使用自带 zk

1、启动zookeeper   当前目录执行:
.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties2、启动kafak   当前目录执行:
.\bin\windows\kafka-server-start.bat .\config\server.properties

使用:

创建topic ： 注意 partitions  分区数，和 replication 副本数
kafka-topics.bat --create --bootstrap-server 127.0.0.1:9092 --topic demp1 --partitions 3 --replication-factor 1 --config cleanup.policy=compact --config retention.ms=500查看主题
kafka-topics.bat --list --bootstrap-server 127.0.0.1:9092删除主题
kafka-topics.bat --delete --bootstrap-server 127.0.0.1:9092 --topic demp1

2.2、linux 安装

参考： https://blog.csdn.net/qq_41463655/article/details/104743410
看安装部分即可，就不重复写了

2.3、数据存储路径

默认: /tmp/kafka-logs
可修改 /config/server.properties 配置文件的 log.dirs=/tmp/kafka-logs 重指定为其他目录即可
存储策略详见第七部分

三、名词解释

名词	中文名	说明
Broker	kafka 服务	kafka 服务器，集群 = 多个Broker
Producer	生产者	向Broker投递消息
Consumer	消费者	从Broker获取消息
Consumer Group	消费者组	同一个组的多个消费者只有一个消费者能获取同一条 topic 消息进行消费多个组的多个消费者能获取同一条 topic 消息进行消费
Topic	主题	存放消息的队列分类，比如邮件、短信、优惠券消息类型
Partition	分区	一个topic 可以指定多个分区，消息会存放到特定分区目的是便于扩展消费者可以集群消费
Replica	副本数据	数据备份, 防灾
Offset	消息位置	标记指定 topic 消费者已经消费到的数据索引位置消费完一条消息后，Offset 索引位置+1
ZooKeeper	分布式协调工具	brokers、topics信息都会存放在zk上

四、springboot 整合 kafka

当前springboot 版本 2.2.2.RELEASE

4.1、pom.xml

         <!-- lombok  生成get/set  --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><!-- spring-mvc --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- springBoot集成kafka --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

4.2、yml

spring:## kafka 配置kafka:consumer:# 服务器地址, 多个逗号分割bootstrap-servers: 127.0.0.1:9092# 指定一个默认的组名# 不使用group的话，启动10个consumer消费一个topic，这10个consumer都能得到topic的所有数据，相当于这个topic中的任一条消息被消费10次。# 使用group的话，连接时带上groupId，topic的消息会分发到10个consumer上，每条消息只被消费1次group-id: kafka2# earliest:  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费# latest:    当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据# none:topic 各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: earliest# key/value的反序列化key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer# 消费者是否自动提交偏移量，默认为trueenable-auto-commit: false# 单次调用 poll方法能够返回的消息数量max-poll-records: 50producer:# key/value的序列化key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer# 批量抓取# batch-size: 65536# 缓存容量# buffer-memory: 524288# 服务器地址# bootstrap-servers: 127.0.0.1:9092

4.3、基础消息投递与接收

点对点模式
topic 相同， groupId 点对点模式作用不大

    @Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;@RequestMapping("/kafka")public void send() {// 主题String topic = Demo1.TOPIC;// 内容String data = "{'userId':'1','name':'wangsong'}";// 发送消息kafkaTemplate.send(topic, data);}/*** 消费者接收消息*/@KafkaListener(topics = Demo1.TOPIC, groupId = "test")public void receive(ConsumerRecord<?, ?> consumer) {System.out.println(" topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}

五、消息投递过程

5.1、消息投递过程

1.生产者投递消息，首先会先将消息投递到kafka客户端缓冲区中，缓冲区满了，在将
该消息投递到Kafka服务器端中。
2.为什么要设计kafka客户端缓冲区，假设在1s内发送1000条消息，如果每次以一条一条发送对kafka服务器端压力非常大，所以在这时候设计一个缓冲区，缓冲区满了，在批量将
消息投递到kafka中，从而减少网络传输的次数。
3.Kafka消费者订阅我们的分区、获取消息，获取消息成功或者失败，消息不会立马被kafka删除。
采用定时删除策略或者当数据满自动清除历史的数据。
log.retention.hours=48 #数据最多保存48小时
log.retention.bytes=1073741824 #数据最多1G
Offset作用：记录消费者消费该分区的进度。

5.2、消费组消费

1.消费者从kafka分区中获取消息，默认的情况下是根据该分组中对应分区已经提交的offset值开始消费。
2.当消费者消费成功之后，默认的情况下会提交该消息的offset值，下次消费者获取消息则从该提交的offset+1开始消费。
3.当分区的数量达到了消费者集群数量，无需再继续新增消费者数量,实现消费者一对一分区消费消息，可以实现高性能。
4.注意，在同一个消费者组中不允许该组中有多个消费者消费同一条消息，多个不同的分组中可以实现不同组中消费者消费同一条消息。

六、消息投递 demo

消息投递 demo 总览

测试demo	测试内容	参数	效果
demo1	基本点对点消息	分区数 1 消费者 1 groupId 随意	消息被消费
demo2	一对多消息	分区数 1 消费者 2 消费者一 groupId=test1 消费者二 groupId=test2	消息被2个消费者同时消费
demo3	多消费者集群	分区数 1 消费者 2 消费者一 groupId=test1 消费者二 groupId=test1	消息只被其中 1个消费者消费
demo4	投递消息分区 + 多消费者集群	分区数 2 消费者 2 消费者一 groupId=test1 消费者二 groupId=test1	消息会以此投递到分区0 和分区 1 消费者1 消费分区 0 数据消费者2 消费分区 1 数据 (分区索引表示)
demo5	多消费者集群消息顺序一致性	分区数 2 消费者 2 消费者一 groupId=test1 消费者二 groupId=test1 指定消息 key	模拟发送1到10 消费者接收消息进行输出不指定 key, A+B 消费者乱序输出 1到10 (异步不确定性) 指定 key 消息只会投递到 1 个分区消息投递到分区0, A消费者进行顺序消费消息投递到分区1, B消费者进行顺序消费
demo6	手动提交 commit offset (默认自动) 消息失败重试	分区 1 消费者 1 groupId 随意 yml enable-auto-commit = false 手动提交 @Bean 配置	消费者使用 ack.acknowledge() 进行提交如果不提交等于没有消费重新启动在再次获取到数据消息处理失败重试3次, 间隔5秒

6.1、demo1 - 基本点对点消息

基本信息，创建主题即可

/*** 创建一个消息   被一个消费者消费* <P>*     topic 相同* </P>* @author wangsong* @email 1720696548@qq.com* @date 2022/6/8 11:54*/
@Component
public class Demo1 {private final static String TOPIC = "topic.demo1";@Beanpublic NewTopic batchTopic() {// 主题  分区数  备份数return new NewTopic(Demo1.TOPIC, 1, (short) 1);}@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 生产者发送消息*/@RequestMapping("/kafka")public void send() {// 主题String topic = Demo1.TOPIC;// 内容String data = "{'userId':'1','name':'wangsong'}";// 发送消息kafkaTemplate.send(topic, data);}/*** 消费者接收消息*/@KafkaListener(topics = Demo1.TOPIC, groupId = "test")public void receive(ConsumerRecord<?, ?> consumer) {System.out.println(" topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}}

6.2、demo2 - 一对多消息

发布订阅，同一个主题多个不同分组的消费者订阅

消息被2个消费者同时消费

/*** 创建一个消息   定义多个消费者 每个消费者能消费* <p>* topic 相同* groupId 不相同* </P>** @author wangsong* @email 1720696548@qq.com* @date 2022/6/8 11:54*/
@Component
public class Demo2 {private final static String TOPIC = "topic.demo2";@Beanpublic NewTopic batchTopic2() {// 主题  分区数  备份数return new NewTopic(Demo2.TOPIC, 1, (short) 1);}@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 生产者发送消息*/public void send() {// 主题String topic = Demo2.TOPIC;// 内容String data = "{'userId':'1','name':'wangsong'}";// 发送消息kafkaTemplate.send(topic, data);}/*** 消费者接收消息*/@KafkaListener(topics = Demo2.TOPIC, groupId = "test1")public void receive(ConsumerRecord<?, ?> consumer) {System.out.println("A消费者 -->   topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}/*** 消费者接收消息*/@KafkaListener(topics = Demo2.TOPIC, groupId = "test2")public void receive2(ConsumerRecord<?, ?> consumer) {System.out.println("B消费者 -->  topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}}

6.3、demo3 - 多消费者集群

发布订阅，同一个主题多个相同分组的消费者订阅（相同分组=集群）

消息只被其中 1个消费者消费

/*** 创建一个消息   定义多个消费者 但只有一个消费者能消费** <P>*     topic 相同*     groupId 相同* </P>** @author wangsong* @email 1720696548@qq.com* @date 2022/6/8 11:54*/
@Component
public class Demo3 {private final static String TOPIC = "topic.demo3";@Beanpublic NewTopic batchTopic3() {// 主题  分区数  备份数return new NewTopic(Demo3.TOPIC, 1, (short) 1);}@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 生产者发送消息*/public void send() {// 主题String topic = Demo3.TOPIC;// 内容String data = "{'userId':'1','name':'wangsong'}";// 发送消息kafkaTemplate.send(topic, data);}/*** 消费者接收消息*/@KafkaListener(topics = Demo3.TOPIC, groupId = "test1")public void receive(ConsumerRecord<?, ?> consumer) {System.out.println("A消费者 -->   topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}/*** 消费者接收消息*/@KafkaListener(topics = Demo3.TOPIC, groupId = "test1")public void receive2(ConsumerRecord<?, ?> consumer) {System.out.println("B消费者 -->  topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}}

6.4、demo4 - 投递消息分区 + 多消费者集群

消息会以此投递到分区0 和分区 1
消费者1 消费分区 0 数据
消费者2 消费分区 1 数据
(分区索引表示)

/*** 创建一个消息。 定义2个分区 和 2个消费者, 每条消息只能有一个消费者能成功消费** <p>* 投递消息依次向分区 0 和 1 投递* A 消费者消费分区 0数据* B 消费者消费分区 1数据* </P>** <P>*     分区 * 2*     消费者 * 2*     topic 相同*     groupId 相同* </P>** @author wangsong* @email 1720696548@qq.com* @date 2022/6/8 11:54*/
@Component
public class Demo4 {private final static String TOPIC = "topic.demo4";@Beanpublic NewTopic batchTopic4() {// 主题  分区数  备份数return new NewTopic(Demo4.TOPIC, 2, (short) 1);}@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 生产者发送消息*/public void send() {// 主题String topic = Demo4.TOPIC;// 内容String data = "{'userId':'1','name':'wangsong'}";// 发送消息kafkaTemplate.send(topic, data);}/*** 消费者接收消息*/@KafkaListener(topics = Demo4.TOPIC, groupId = "test1")public void receive(ConsumerRecord<?, ?> consumer) {System.out.println("A消费者 -->   topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}/*** 消费者接收消息*/@KafkaListener(topics = Demo4.TOPIC, groupId = "test1")public void receive2(ConsumerRecord<?, ?> consumer) {System.out.println("B消费者 -->  topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}}

6.5、demo5 - 多消费者集群消息顺序一致性

定义key ，消息会全部投到一个分区，一个分区的消息会落到同一个消费者进行顺序消息

/*** 消息一致性问题* 创建一个消息。 定义2个分区 和 2个消费者, 每条消息只能有一个消费者能成功消费** <p>* 投递消息依次向分区 0 和 1 投递* A 消费者消费分区 0数据* B 消费者消费分区 1数据* <p>* 处理方式： 定义 key 即可* // 不设置key A消费者+ B消费者 会同时获取消息，导致输出内容中 for的 i 值是乱序输出的* // 设置key 后, 该消息只会向一个分区中投递消息, 且一个分区的数据相同groupId分组下只会被一个消费者消费* </P>** <p>* 分区 * 2* 消费者 * 2* topic 相同* groupId 相同* </P>** @author wangsong* @email 1720696548@qq.com* @date 2022/6/8 11:54*/
@Component
public class Demo5 {private final static String TOPIC = "topic.demo5";@Beanpublic NewTopic batchTopic5() {// 主题  分区数  备份数return new NewTopic(Demo5.TOPIC, 2, (short) 1);}@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 生产者发送消息*/public void send() {// 主题String topic = Demo5.TOPIC;// 内容for (int i = 0; i < 10; i++) {String data = i + "  {'userId':'1','name':'wangsong'}";// 可以保证消息顺序一致性// 不设置key A消费者+ B消费者 会同时获取消息，导致输出内容中 for的 i 值是乱序输出的// 设置key 后, 该消息只会向一个分区中投递消息, 且一个分区的数据相同groupId分组下只会被一个消费者消费String key = "demo5";// 发送消息kafkaTemplate.send(topic, key, data);}}/*** 消费者接收消息*/@KafkaListener(topics = Demo5.TOPIC, groupId = "test1")public void receive(ConsumerRecord<?, ?> consumer) {System.out.println("A消费者 -->   topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}/*** 消费者接收消息*/@KafkaListener(topics = Demo5.TOPIC, groupId = "test1")public void receive2(ConsumerRecord<?, ?> consumer) {System.out.println("B消费者 -->  topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());}}

6.6、demo6 - 手动提交 commit offset + 消息重试

注意配置

1、需配置 @Bean 设置为手动提交模式，即下方注释代码
2、yml 配置 enable-auto-commit 设置为 false | spring.kafka.consumer.enable-auto-commit = false
3、消费者使用 ack.acknowledge() 提交，表示成功消费

/*** 消息手动 commit offset** <p>* 1。需配置 @Bean 设置为手动提交模式* 2、yml 配置 enable-auto-commit 设置为 false* 3、当不提交时 ack.acknowledge(), 每次重启消费者服务都会拉取到消息，没提交表示该没有被成功消费* </P>** @author wangsong* @email 1720696548@qq.com* @date 2022/6/8 11:54*/
@Component
@Slf4j
public class Demo6 {private final static String TOPIC = "topic.demo6";@Beanpublic NewTopic batchTopic6() {// 主题  分区数  备份数return new NewTopic(Demo6.TOPIC, 1, (short) 1);}// 配置手动提交offset@Beanpublic KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);factory.getContainerProperties().setPollTimeout(1500);// 消息是否或异常重试次数3次 (5000=5秒  3=重试3次)// 注意： 没有配置配置手动提交offset,不生效, 因为没有配置手动提交时消息接受到就会自动提交，不会管程序是否异常SeekToCurrentErrorHandler seekToCurrentErrorHandler = new SeekToCurrentErrorHandler((consumerRecord, e) -> {log.error("消费消息异常.抛弃这个消息,{}", consumerRecord.toString(), e);}, new FixedBackOff(5000, 3));factory.setErrorHandler(seekToCurrentErrorHandler);// 配置手动提交offsetfactory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);return factory;}@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 生产者发送消息*/public void send() {// 主题String topic = Demo6.TOPIC;// 内容String data = "{'userId':'1','name':'wangsong'}";// 发送消息kafkaTemplate.send(topic, data);}/*** 消费者接收消息*/@KafkaListener(topics = Demo6.TOPIC, groupId = "test1")public void receive(ConsumerRecord<?, ?> consumer, Acknowledgment ack) throws Exception {System.out.println("A消费者 -->   topic名称:" + consumer.topic()+ " , key:" + consumer.key()+ " , 分区位置:" + consumer.partition()+ " , 下标:" + consumer.offset()+ " , 内容:" + consumer.value());throw new Exception("程序错误");//ack.acknowledge();}
}

七、kafka 消息存储策略

7.1、存储机制

Kafka mq（日志存储系统）服务端存储消息的时候，不管该消息是否有消费成功，最终该消息不会立即被删除。

1、基于时间，默认配置是168小时（7天）。
2、基于大小，默认配置是1073741824 (1GB)。

7.2、消息存储位置

默认: /tmp/kafka-logs
可修改 /config/server.properties 配置文件的 log.dirs 参数指定

7.3、消息存储方式

每个分区都会有一个文件，一个分区则只有一个

topic 存储方式
如图

底层持久化我们消息设计到：segment

1、在一个分区中，会将一个大的分区拆分 n多个不同小 segment文件
2、在每个segment中会有.index、.log .timeindex 等文件
3、每个 segment file 默认的大小是为500MB，每次达到容量后，就往后在创建一个新的segment file

.index -----消息偏移量索引文件
.log -----消息持久化内容(日志文件)
.timeindex -----时间戳索引文件

7.4、segment file 命名规则

1、每个segment file也有自己的命名规则，每个名字有20个字符，不够用0填充，每个名字从0开始命名，
2、下一个segment file文件的名字就是，上一个segment file中最后一条消息的索引值。
3、在.index文件中，存储的是key-value格式的，key代表在.log中按顺序开始顺序消费的offset值，value代表该消息的物理消息存放位置。但是在.index中不是对每条消息都做记录，它是每隔一些消息记录一次，避免占用太多内存。即使消息不在index记录中，在已有的记录中查找，范围也大大缩小了。
4、每个 log 文件的大小是一样的，但是存储的 message 数量是不一定相等的（每条的 message 大小不一致）。文件的命名是以该 segment 最小 offset 来命名的，如 000.index 存储offset为0~368795的消息，kafka就是利用分段+索引的方式来解决查找效率的问题

如下：一个分区中拆分n多个不同的小的segment 文件

Segment0 容量满的情况下 500条消息创建offset 500 (具体名称看规则 )
Segment500 容量满的情况下 500条消息创建 offset 1000 (具体名称看规则 )
Segment1000

7.5、如何通过offset 查询数据的

查询是根据 offset 索引进行查询, 第一步截取 index 文件名后几位, 判断数据在那个 segment 文件中

现在假设数据都早 00000000000000000.index .log 中
.index中不是对每条消息都做 offset 索引记录 (详见 7.4.3)

数据存储如下图:

理想状态如图：获取索引为 1、3、6、9 的数据，存在索引，通过索引直接获取到返回就可以了

非理想状态下如图：获取索引为 2,、4、5、7 的数据，不存在索引，使用二分法快速定位数据所在位置

其他: index中实际 offset 数据间隔可能在几百，二分折半算法查找快

7.6、命令查询索引文件以及日志数据

需在 kafka 命令目录下指定文件执行

kafka-run-class.bat kafka.tools.DumpLogSegments --files 00000000000000000000.log
kafka-run-class.bat kafka.tools.DumpLogSegments --files 00000000000000000000.index

八、副本数据 (集群)

8.1、副本在kafka中的作用

所谓的副本机制（Replication）,也可以称之为备份机制，通常是指分布式系统在多台互联网的机器上保存相同的数据拷贝。

根据 kafka副本机制的定义，同一个分区下的所有副本保存有相同的消息序列，这些副本分散的保存在不同的Broker上，从而能够对抗部分Broker宕机带来的数据不可用。

主从
leader 主 | Follwer 从
这里请求访问到那个Broker ，那个Broker 都等同于 leader

消息同步机制
leader 节点消息接受消息后持久化到硬盘并同步给其他两个 Follwer 的 Broker

主从机制 (ISR 机制)

假设 Broker1 宕机了，会重新选举Broker1 的 leader 节点， ISR 集合中会存储所有 Broker 节点数据，从 ISR 集合获取存活的第一个Broker服务为当前宕机服务的 leader 节点进行继续使用

ISR：活跃的副本节点集合，从集合中重新选举

8.2、副本ack 机制

Ack=0
延迟概率比较低、有可能导致消息会丢失。
生产者不需要mq服务器ack确认。
Ack=1 (建议)
延迟概率比较适中，避免消息丢失概率比较低。
生产者投递消息只需要我们 leader 能够将该消息持久化成功，在返回ack状态码给我们生产者角色。
如果leader 宕机之后，leader 没有及时将该消息发送其他的Follwer 节点
会导致数据会丢失。
Ack=-1
延迟概率比较高，避免消息不丢失
生产者投递消息需要所有的 leader 和 Follwer 全部持久化成功，在返回ack给生产者角色。

注意：
生产者投递消息给 leader 和 Follwer 持久化成功，但是这时候突然我们的
leader 节点宕机，宕机之后leader 无法告诉结果给我们生产者，生产者误以为
该消息投递失败，则会不断做重试，重试的过程中就会再剩余的 Follwer 节点中查找一个新的 leader 节点，投递该相同的消息消息

如何处理：消费者根据全局id 保证幂等性问题

8.3、副本宕机重启同步机制

Leo：当前副本数据中最后的一个offset值（最大offset） -> 15 9 11
HW：消费者能够获取到最大offset值。

洌1：

副本3宕机重启，
第一步，清除HW=9 后的值 10 和 11，
在同步大于当前副本3 offset =11 的副本1的 offset =15 的值（同步10到15）

洌1：

副本1 宕机重启，
第一步，清除HW=9 后的值 10 到 15
在同步大于当前副本1 offset =15 的值（没有），当前副本节点到9结束，可能丢失数据

8.4、kafka 控制器

kafka 控制器只存在于一个 broker 中, 依托于 zk 实现和选举

1、kafka 控制器作用

1.主题管理（创建、删除、增加分区）
这里的主题管理，就是指控制器帮助我们完成对 Kafka 主题的创建、删除以及分区增加的操作。换句话说，当我们执行kafka-topics 脚本时，大部分的后台工作都是控制器来完成的。关于 kafka-topics 脚本，我会在专栏后面的内容中，详细介绍它的使用方法。
2.分区重分配
分区重分配主要是指，kafka-reassign-partitions 脚本（关于这个脚本，后面我也会介绍）提供的对已有主题分区进行细粒度的分配功能。这部分功能也是控制器实现的。
3.集群成员管理
自动检测新增 Broker、Broker 主动关闭及被动宕机。这种自动检测是依赖于前面提到的 Watch 功能和 ZooKeeper 临时节点组合实现的。比如，控制器组件会利用Watch 机制检查 ZooKeeper 的 /brokers/ids 节点下的子节点数量变更。目前，当有新 Broker 启动后，它会在 /brokers 下创建专属的 znode 节点。一旦创建完毕，ZooKeeper 会通过 Watch 机制将消息通知推送给控制器，这样，控制器就能自动地感知到这个变化，进而开启后续的新增 Broker 作业。
侦测 Broker 存活性则是依赖于刚刚提到的另一个机制：临时节点。每个 Broker 启动后，会在 /brokers/ids 下创建一个临时 znode。当 Broker 宕机或主动关闭后，该 Broker 与 ZooKeeper 的会话结束，这个 znode 会被自动删除。同理，ZooKeeper 的 Watch 机制将这一变更推送给控制器，这样控制器就能知道有 Broker 关闭或宕机了，从而进行“善后”。
4.数据服务
控制器的最后一大类工作，就是向其他 Broker 提供数据服务，控制器上保存了最全的集群元数据信息，其他所有 Broker 会定期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据。

2、kafka 控制器选举

当控制器发现一个 broker 离开集群（通过观察相关 ZooKeeper 路径），控制器会收到消息：这个 broker 所管理的那些分区需要一个新的 Leader。控制器会依次遍历每个分区，确定谁能够作为新的 Leader，然后向所有包含新 Leader 或现有 Follower 的分区发送消息，该请求消息包含谁是新的 Leader 以及谁是 Follower 的信息。随后，新的 Leader 开始处理来自生产者和消费者的请求，Follower 用于从新的 Leader 那里进行复制。

举例
假设现在：broker0 宕机之后，控制器通过zookeeper 临时节点事件通知得到感
通知给其他的 broker 更新对应副本信息状态（ISR集合列表）

九、kafka 参数配置

参数	默认	配置说明
buffer.memory	32M	缓冲区大小
batch.size	16kb	缓冲区分批大小
max.request.size		服务器接受消息大小限制
retries		重试次数
retries.backoff.ms		间隔时间
acks		消息确认机制 0 1 - 1
分区数

待定
可查阅 spring-kafka 源码中的 KafkaProperties.java 文件参数