大家好，又见面了，我是你们的朋友全栈君。

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故障处理细节

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值，只保留最后一个版本。

• log.cleanup.policy = compact 所有数据启用压缩策略

这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。

高效读写数据

• Kafka 本身是分布式集群，可以采用分区技术，并行度高。
• 读数据采用稀疏索引，可以快速定位要消费的数据。
• 顺序写磁盘（追加到文末）
• 页缓存 + 零拷贝技术

零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。

PageCache页缓存：Kafka重度依赖底层操作系统提供的PageCache功 能。当上层有写操作时，操作系统只是将数据写入PageCache。当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。

消费者

消费者工作流程

总体工作流程图

消费者组原理

• 消费者组是由多个消费者组成，组内所有消费者的groupid相同。
• 每个分区的数据只能由消费者组中的一个消费者消费。
• 消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

分区的分配以及再平衡

kafka有四种主流的分区分配策略：Range、RoundRobin、Sticky、CooperativeSticky，默认是Range+CooperativeStricky，可以同时使用多个分区分配策略。

offset 位移

• kafka0.9版本之前，consumer默认将offset保存在zookeeper中。（大量消费时与zookeeper的交互成为性能瓶颈）
• 从0.9版本开始，consumer默认将offset保存在kafka一个内置的topic中，该topic为 __consumer_offsets

---

本文总结自尚硅谷视频：kafka，仅作个人学习笔记使用。