大数据|kafka--核心技术篇大数据|kafka

kafka生产者生产者消息发送流程发送原理
在消息发送的过程中，涉及到了两个线程——main 线程和== Sender 线程==。在 main 线程中创建了一个双端队列 RecordAccumulator（32M内存）。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

文章图片

拦截器：对数据进行加工处理（可选）。
序列化器：将对象的状态信息转换为可以存储或传输的形式。
分区器：决定数据发送到哪个分区。
batch.size：只有数据积累到batch.size之后，sender才会发送数据（默认16k）。
linger.ms：如果数据迟迟未达到batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。

sender从RecordAccumulator拉取数据以broker为key，要发往同一个broker的数据在一个请求队列中，默认每个broker节点最多缓存5个请求。
broker接收数据时的应答机制acks：

0：生产者发送过来的数据，不需要等数据落盘应答。
1：生产者发送过来的数据，Leader收到数据后应答。
-1（all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。-1和all等价。

生产者重要参数列表

参数名称	描述
bootstrap.servers	生产者连接集群所需的 broker 地址清单。例如s1:9092,s2:9092,s3:9092，可以设置 1 个或者多个，中间用逗号隔开。注意这里并非需要所有的 broker 地址，因为生产者可以从给定的 broker里查找到其他 broker 信息。
key.serializer 和 value.serializer	指定发送消息的 key 和 value 的序列化类型。一定要写全类名。
buffer.memory	RecordAccumulator 缓冲区总大小，默认 32m。
batch.size	缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
linger.ms	如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。
acks	0：生产者发送过来的数据，不需要等数据落盘应答。 1：生产者发送过来的数据，Leader 收到数据后应答。 -1（all）：生产者发送过来的数据，Leader+和 isr 队列里面的所有节点收齐数据后应答。默认值是-1，-1 和all 是等价的。
max.in.flight.requests.per.connection	允许最多没有返回 ack 的次数，默认为 5，开启幂等性要保证该值是 1-5 的数字。
retries	当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。
retry.backoff.ms	两次重试之间的时间间隔，默认是 100ms。
enable.idempotence	是否开启幂等性，默认 true，开启幂等性。
compression.type	生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。

生产者分区 kafka分区好处

便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。
提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。

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

batch.size：批次大小，默认16k
linger.ms：等待时间，修改为5-100ms
compression.type：压缩snappy
RecordAccumulator：缓冲区大小，修改为64m
根据生产环境来调整上述参数，以达到适宜的数据拉取速度。

数据可靠性
取决于应答级别

acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；

在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。
数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2
数据去重

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。
精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2）。
事务：Producer 在使用事务功能前，必须先自定义一个唯一的 transactional.id。有了 transactional.id，即使客户端挂掉了，它重启后也能继续处理未完成的事务

kafka broker kafka broker工作流程 zookeeper存储的kafka信息

/kafka/brokers/ids [0,1,2] 记录有哪些服务器
/kafka/brokers/topics/first/partitions/0/state
{“leader”:1 ,“isr”:[1,0,2] } 记录谁Leader，有哪些服务器可用
/kafka/controller
{“brokerid”:0} 辅助选举Leader

kafka broker总体工作流程

文章图片

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

broker重要参数

参数名称	描述
replica.lag.time.max.ms	ISR 中，如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值，默认 30s。
auto.leader.rebalance.enable	默认是 true。自动 Leader Partition 平衡。
leader.imbalance.per.broker.percentage	默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。
leader.imbalance.check.interval.seconds	默认值 300 秒。检查 leader 负载是否平衡的间隔时间。
log.segment.bytes	Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分成块的大小，默认值 1G。
log.index.interval.bytes	默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。
log.retention.hours	Kafka 中数据保存的时间，默认 7 天。
log.retention.minutes	Kafka 中数据保存的时间，分钟级别，默认关闭。
log.retention.ms	Kafka 中数据保存的时间，毫秒级别，默认关闭。
log.retention.check.interval.ms	检查数据是否保存超时的间隔，默认是 5 分钟。
log.retention.bytes	默认等于-1，表示无穷大。超过设置的所有日志总大小，删除最早的 segment。
log.cleanup.policy	默认是 delete，表示所有数据启用删除策略；如果设置值为 compact，表示所有数据启用压缩策略。
num.io.threads	默认是 8。负责写磁盘的线程数。整个参数值要占总核数的 50%。
num.replica.fetchers	副本拉取线程数，这个参数占总核数的 50%的 1/3
num.network.threads	默认是 3。数据传输线程数，这个参数占总核数的50%的 2/3 。
log.flush.interval.messages	强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。
log.flush.interval.ms	每隔多久，刷数据到磁盘，默认是 null。一般不建议修改，交给系统自己管理。

节点服役和退役服役新节点

以上线broker3举例

1）新节点配置

安装jdk
安装kafka
配置broker.id、log.dirs、zookeeper.connect
配置环境变量
启动zookeeper
启动kafka

2）执行负载均衡操作（在现役节点）（1）创建一个要均衡的主题。

vim topics-to-move.json{ "topics": [ {"topic": "first"} ], "version": 1 }

（2）生成一个负载均衡的计划。

broker-list里添加3

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

得到当前的副本分配和计划的副本分配，如下所示：

Current partition replica assignment
{“version”:1,“partitions”:[{“topic”:“first”,“partition”:0,“replicas”:[0,2,1],“log_dirs”:[“any”,“any”,“any”]},{“topic”:“first”,“partition”:1,“replicas”:[2,1,0],“log_dirs”:[“any”,“any”,“any”]},{“topic”:“first”,“partition”:2,“replicas”:[1,0,2],“log_dirs”:[“any”,“any”,“any”]}]}

Proposed partition reassignment configuration
{“version”:1,“partitions”:[{“topic”:“first”,“partition”:0,“replicas”:[2,3,0],“log_dirs”:[“any”,“any”,“any”]},{“topic”:“first”,“partition”:1,“replicas”:[3,0,1],“log_dirs”:[“any”,“any”,“any”]},{“topic”:“first”,“partition”:2,“replicas”:[0,1,2],“log_dirs”:[“any”,“any”,“any”]}]}

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

vim increase-replication-factor.json

内容为刚才生成的预计负载均衡计划（Proposed partition reassignment configuration）
（4）执行副本存储计划。

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

（5）验证副本存储计划。

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

退役旧节点

以退役broker3举例

1）执行负载均衡操作（1）创建一个要均衡的主题，方法同上。
（2）创建执行计划

borker-list里去掉3

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

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

vim increase-replication-factor.json

内容为刚得到的副本存储计划，注意文件名的重复性。
（4）执行副本存储计划，方法同上。
（5）验证副本存储计划，方法同上。
2）执行停止命令在broker3上执行停止命令：

bin/kafka-server-stop.sh

kafka副本副本基本信息

Kafka 副本作用：提高数据可靠性。
Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。
Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据。
Kafka 分区中的所有副本统称为 AR（Assigned Repllicas）。

Leader选举流程
Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。
Controller 的信息同步工作是依赖于 Zookeeper 的。

文章图片

Leader和Follower故障处理细节

【大数据|kafka--核心技术篇】LEO（Log End Offset）：每个副本的最后一个offset，LEO其实就是最新的offset + 1。
HW（High Watermark）：所有副本中最小的LEO 。

Follower故障（1） Follower发生故障后会被临时踢出ISR
（2）这个期间Leader和Follower继续接收数据
（3）待该Follower恢复后，Follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向Leader进行同步。
（4）等该Follower的LEO大于等于该Partition的HW，即Follower追上Leader之后，就可以重新加入ISR了。
Leader故障（1） Leader发生故障之后，会从ISR中选出一个新的Leader
（2）为保证多个副本之间的数据一致性，其余的Follower会先将各自的log文件高于HW的部分截掉，然后从新的Leader同步数据。
分区副本分配
4broker16分区3副本分配情况

文章图片

手动调整分区副本存储在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储。
（1）创建副本存储计划。
以所有副本都制定存储到broker0、broker1中为例：

vim increase-replication-factor.json{ "version":1, "partitions":[{"topic":"three","partition":0,"replicas":[0,1]}, {"topic":"three","partition":1,"replicas":[0,1]}, {"topic":"three","partition":2,"replicas":[1,0]}, {"topic":"three","partition":3,"replicas":[1,0]}] }

（2）执行副本存储计划。

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

（3）验证副本存储计划

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

文件存储文件存储机制

文章图片

Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。

.log 日志文件，默认值 1G。
.index 偏移量索引文件，为稀疏索引，大约每往log文件写入4kb数据，会往index文件写入一条索引。Index文件中保存的offset为相对offset，这样能确保offset的值所占空间不会过大，因此能将offset的值控制在固定大小
.timeindex 时间戳索引文件

顺序写入+稀疏索引保障了kafka的高吞吐量性能
文件清理策略
Kafka 中默认的日志保存时间为 7 天，可以通过调整如下参数修改保存时间。

log.retention.hours，最低优先级小时，默认 7 天。
log.retention.minutes，分钟。
log.retention.ms，最高优先级毫秒。
log.retention.check.interval.ms，负责设置检查周期，默认 5 分钟。

超过保存时间有两种处理方式，一种是delete删除，另一种是compact压缩。
delete日志删除

log.cleanup.policy = delete 所有数据启用删除策略
log.retention.bytes 删除的大小限制，默认等于-1，表示无穷大。
（1）基于时间：默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。
（2）==基于大小：默认关闭。==超过设置的所有日志总大小，删除最早的 segment。

如果一个 segment 中有一部分数据过期，一部分没有过期，则会等待所有数据过期后再删除。
compact日志压缩 compact日志压缩：对于相同key的不同value值，只保留最后一个版本。