《深入理解kafka》阅读笔记

常用命令

• 创建topic

  ./kafka-topics.sh  --bootstrap-server localhost:9092
  --topic topic-create 
  --create 
  --partitions 4
  --replication-factor 2
  

• 查看所有的topic

  ./kafka-topics.sh --bootstrap-server localhost:9092
  --list
  

• 查看topic详情

  ./kafka-topics.sh --bootstrap-server localhost:9092
  --topic topic-create
  --describe
  

• 查看所有的消费者组

  ./kafka-consumer-group.sh --bootstrap-server localhost:9092 
  --list
  

• 查看消费者组详情

  ./kafka-consumer-group.sh --bootstrap-server localhost:9092
  --describe
  --group group_name
  
  

• 生产msg

  kafka-console-producerd.sh
  --broker-list localhost:9092
  --topic topic-create
  

• 消费msg

  kafka-console-consumer.sh
  --bootstrap-server localhost:9092
  --topic topic-create
  

• 查看consumer消息

  ./kafka-console-consumer.sh --bootstrap-server localhost:9092 
  --topic __consumer_offsets 
  --formatter 'kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter' 
  --from-beginning
  --partition 10
  

  --partition的值不是固定的，Math.abs(groupID.hashCode()) % numPartitions计算得来的

协议

• kafka的消息格式:

CRC: 校验码

属性位：0（无压缩）、1（GZIP）、2（Snappy）和3（LZ4）。

KEY： 消息键，对消息做partition时使用，即决定消息被保存在某topic下的哪个partition。

Value：消息体，保存实际的消息数据。

Timestamp：消息发送时间戳，用于流式处理及其他依赖时间的处理语义。如果不指定则取当前时间

• kafka的复制机制:

kafka 的复制机制既不是完全的同步复制，也不是单纯的异步复制。同步复制要求所有能工作的 follower 副本都复制完，这条消息才会被确认为已成功提交，这种复制方式极大地影响了性能。而在异步复制方式下，follower 副本异步地从 leader 副本中复制数据，数据只要被 leader 副本写入就被认为已经成功提交。在这种情况下，如果 follower 副本都还没有复制完而落后于 leader 副本，突然 leader 副本着机，则会造成数据丢失。 Kafka 使用的ISR 的方式则有效地权衡了数据可靠性和性能之间的关系。

生产者

• 生产者客户端

消息从主线程后会到消息累加器，消息累加器为每一个分区维护了一个deque，该双端队列存储着不同的Batch，Batch是用来存储一批消息以提升发送效率的。

然后这些Batch就会被Sender线程读取，包装成一个Request发送出去，并维护了一个InFlightRequest队列,该队列记录Request的Response状态

生产者客户端是与具体 broker 节点建 的连接，也就是 具体的 broker 节点发送消息，而并不关心消息属于哪一个分区；向哪个分区中发送哪些消息，客户端底层会做一个应用逻辑层面到网络 I/O 层面的转换。

• 生产者参数

  acks:

  0: tcp层面上的消息发送成功就返回ack

  1(默认): leader将消息成功写盘就返回ack

  -1/all : ISR所有副本都将消息写盘后才返回ack

​ retires ：

​ 重试次数，默认为0

​ retry.backoff.ms:

​ 重试间隔，默认100ms

​ max.in.flight.requests.per.connection:

​ 每个connection中的未得ack的request的数量，要使用kafka的分区顺序性，需将此值设成1，否则第一个request发送失败后，第二个request发送完成了这时才会发送第一个request.

消费者

• 消费者位移

  新建立的消费者消费的消息offset由auto.offset.reset设定，有以下值:

  latest(默认):从分区末尾开始消费

  earlist:从分区开头开始消费

  none: 提示错误，报异常

​ 更改offset也可以用seek()方法, seek()方法改变的offset超过分区数据尾，将会收到broker的reset消息，毕更具auto.offset.reset进行重置。

• 再均衡监听器

  发生在消费组变更的时候，此时会导致部分consumer未commit而partition被分配给其它的消费者，进而重复消费，解决方法：

  使用再均衡监听器，实现回调OnPartitionsRevoked()， 使用该方法进行提前commit可以避免问题

• 多线程消费

  多线程消费的模型有两种：

  1.起N个线程，每个线程使用thread_local保存一个局部的consumer，让这个consumer去poll消息，拉下来后消费处理，这种就很想one loop per thread

  2.起N个线程设置consumer去poll消息，把这些消息提交任务给handle线程池去处理这些消息，这很像muduo或者libevent中的的one loop in per thread + thread pool的模式

​ 线程池消费弊端

​ 异步线程之间的消费完成后的commit顺序无法保证，先提交较大的会掩盖offset小的消息，可以将整个消息队列使用一个滑动窗口处理，消费上必须满足左端窗口完成后方能向右边滑动，滑动时进行一次commit，同时进行了任务数的控制以及commit上的控制

​ 如果某个消息无法消费完成，就会造成窗口悬停，为了使窗口能继续滑动，需要设置一个与之，当startOffset悬停一段时间后需要对失败消息，将之放转入重试队列，如果还失败就放入死信队列，业务中可以将死信做持久化存储下来，然后继续消费下一条消息以保证整体消息进度的合力推进

主题与分区

• 创建topic

  bin/kafka-topics.sh 
  --zookeeper localhost:2181/kafka 
  --create 
  --topic topic-create 
  --partitions 4
  --replication-factor 2
  

  注意其中的--replication-factor代表的是副本的个数(是包含主动分区的)

• 手动分区

  bin/kafka-topics.sh
  --zookeeper localhost:2181/kafka
  --create topic-create
  --replica--assignment 2:0,0:1,1:2,2:1
  

  replica-assignment后面跟着的参数释意为：

  [broker_id_for_part1_replica1:broker_id_for_part1_replica2,

  broker_id_for_part2_replica1:broker_id_for_part2_replica2，...]

  其不允许同一个分区的副本重复入 0:1, 1:1，跳过一个分区的要是不被允许

• 创建topic的限制

  创建同名topic会报错，也可以创建topic的时候带上--if-not-exists

  topic名字不能包含.或_特殊字符

• topic创建的流程

  向zookeeper发送请求，把分区和副本方案写入/config/topics，把topic的配置写入/config/topics中

  kafka同步zookeeper的数据，开始执行

• 查询topic

bin/kafka-topics.sh 
--zookeeper localhost:2181/kafka 
--describe 
--topic topic-create 

• 一些重要的配置参数

  topic中所有配置参数都在broker的配置参数中有所对应，如果topic没有设置，就沿用broker中的

  topic	broker	释义
  cleanup.policy	log.cleanup.policy	日志压缩策略。默认值为 delete ，还可以配compact
  compression.type	compression.type	消息的压缩类型。默认值为 producer ,保持生产生的压缩类型，还可以配置为: uncompressed、snappy、 lz4、gzip
  delete.retension.ms	log.cleaner.delete.retention.ms	被标识为删除的数据能够保留多久 默认值：8640000， 即1天
  file.delete.delay.ms	log.segment.delete.delay.ms	清理文件之前可 以等待多长时间，默认值为，60000 ，即 1分钟
  flush .messages	log.flush.interval.messages	需要收集多少消息才会将它 强制刷新磁盘，默认值为LONG.MAX_VALUE， 即让操作系统决定
  flush.ms	log.flush.interval.ms	需要等待多久才会将消息刷到磁盘中，默认值为：LONG.MAX_VALUE，即让操作系统决定
  follower.replication.throttled.replicas	follower.replication.throttled.replicas
  index.interval. bytes	log.index.interval.bytes
  leader.replication.throttled.replicas	leader.replication.throttled.replicas
  max. message. bytes	message. max. bytes	消息的最大字节数，默认值为 1000012
  message. format. version	log.message forrηat.version
  message. timestamp .difference.max.ms	log.message. timestamp.difference.max.ms
  message. timestamp. type	log.message. timestamp.type	消息的时间戳类型:CreateTime, LogAppendTime
  min.cleanable.dirty.ratio	log.cleaner.min.cleanable.ratio
  min.compaction.lag.ms	log.cleaner.min.compaction.lag.ms
  min. insync. replicas	min.insync.replicas	分区ISR中的最小数
  preallocate	log.preallocate	创建日志Segment是否要预留空间: true/false
  retention.bytes	log.retention.bytes	分区中保存的消息总量，-1（无限制）
  retention.ms	log.retention.ms	数据保持时间，默认为一周，-1则表示无限制
  segment.bytes	log.segment.bytes	日志Segment最大值，默认1GB
  segment.index.bytes	log.index.size.max.bytes	Segment的index文件最大值，默认为10MB
  segment.jitter.ms	log.roll.jitter.ms
  segment.ms	log.roll.ms
  unclean.leader.election.enable	unclean.leader.election.enable	是否允许从非ISR中选取leader

• 删除topic

  bin/kafka-topics.sh --zookeeper localhost:2181/kafka --delete --topic topic-unknown --if-exists
  

  删除topic的型为是在zookeeper中的/admin/delete_topics路上下创建一个于待删除主题桶名的节点，以标记该主题为待删除状态，与创建主题相同的是，真正删除主题的动作也是由kafka的控制器负责完成。

• kafka-topics.sh脚本中的参数

  参数名称	释义
  alter	修改主题，分区数以及主题的配置
  config	设置topic的参数
  create	创建topic
  delete	删除topic
  delete-config	删除topic被覆盖的配置
  describe	查看topic的详细
  disable-rack-aware	创建topic不考虑几级啊的信息
  help	打印帮助信息
  if-exists	删除主题时使用
  if-not-exists
  list
  partitions
  replica-assignment
  replication-factor
  topic
  topics-with-overrides
  unavailable-partitions
  under-replicated-partitions
  zookeeper

副本

• 优先副本与均衡

  

  kafka如果保证分区的均衡性？

  分区负载的均衡主要体现在leader上，Kafka引入了优先副本(preferred replica)的概念， 优先副本是指分区的第一个集合副本(Replicas)，大多情况下，优先副本即是分区的leader； kafka通过确保所有主题优先副本在kafka中均匀分布就保证了leader的均匀分布。

  kafka中提供auto.leader.rebalance.enable用于开启分区leader自平衡，其会使kafka的控制器启动一个定时任务轮询所有的broker节点，超过某比值便会重新选举达到优先副本的平衡，使用优先副本进行更新，但生产环境慎用，因为topic在broker中的均衡不代表服务负载在整个集群中的平衡，

• 分区重分配

  节点变化的时候，集群如何平衡topic?

  集群添加的broker只有新创建的topic才会分配到该节点，可使用kafka-reassign-partitions.sh执行脚本进行重分配，重分配的方案可以手工定义也可以通过自动生成，重分配先通过控制器为每个分区添加新副本，新副本将从分区的leader副本中复制所有的数据，复制完成之后，控制器将从旧副本中从副本清单里移除

• 复制限流

  如何控制复制时的网络流量？

  follower.replication.throttled.rate与leader.replication.throttled.rate是broker上控制复制限流的两个参数，前者用于设置follower副本复制的速度，后者用于设置leader的传输速度，可以使用kafka-config.sh脚本对这两个参数进行管理达到动态限流

  bin/kafka-configs.sh --zookeeper localhost:2181/kafka --entity-type brokers --entity-name 1 --alter --add-config 
  follower.replication.throttled.rate=1024, leader.replication.throttled.rate=1024
  

  keader.replication.throttled.replicas和follower.replication.throttled.replicas是topic上控制leader和follower传输速度的两个配置参数

• 性能测试工具

  kafka-producer-perf-test.sh和kafka-consumer-perf-test.sh可以分别用来做生产者和消费者的测试

  bin/kafka-producer-perf-test.sh
  --topic topic-1
  --num-records 1000000 // 消息数
  --record-size 10000	  // 消息大小
  --throughput -1		  // 网络限流，-1的时候不限流
  --producer-props 	  // 设置生产者的一些配置
  boostrap.servers=localhost:9092
  acks=1
  

  bin/kafka-consumer-perf-test.sh
  --topic topic-1
  --messages 1000000
  --broker-list localhost:9092
  

  于生产者而言，每一个分区的数据写入是并行化的；kafka只允许单个分区中的消息被一个消费者线程消费，一个消费组的并行度完全依赖于所消费的分区数

• 考量因素

  如何保证消息的分区和顺序性？

  kafka通过消息的key计算出消息讲消息写入具体的分区，据由相同key的数据可以写入同一个分区。如果要求主题中的消息保证顺序性，可以讲创建主题时的分区数设置为1

日志存储

• 数据存储布局s

  

  一个topic是由多个partitions组成，一个partitions使用一个log文件夹去存储数据，为了加强数据管理(过期删除)，每个log文件夹下的数据按大小(1G)切分成一个个segment文件夹，每个segment中， log文件用来存数据，.index文件用来索引log中的数据；

  .log文件和.index文件按照固定的20位数字，记录文件第一条数字的偏移，比如000000000000000000.log(0开始)

• offset存储布局

  Kafka 0.9 版本之前，consumer 默认将 offset 保存在 Zookeeper 中，从 0.9 版本开始， consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中，该 topic 为__consumer_offsets。 可以执行以下命令获取

  bin/kafka-console-consumer.sh
  --topic __consumer_offsets 
  --zookeeper hadoop102:2181 
  --formatter "kafka.coordinator.GroupMetadataManager$OffsetsMessageFormatter" 
  --consumer.config config/consumer.properties
  --from-beginning
  

• 消息格式

  

0.10版本到0.11版本之前使用该格式，timestamp由broker端参数log.message.timestamp.type配置，默认为CreateTime, 生产者创建该消息的时间戳，如果创建该Record时未指定消息的时间戳KafkaProducer发送该消息前也会自动加上。

kafka中是将多个消息当作一个消息集(message set)， 消息集是传输和压缩保存的基础单位，消息压缩时将整个消息集进行压缩作为内层消息（inner message）， 内层消息整体作为外层消息（wrapper message）的value，其结构如下:

每个从生产者发出的消息集中的消息offset都是从0开始的，对offset的转换会在服务端进行，外层消息保存了内存消息最后一条小的的绝对offset, 如下图，内层消息的最后一条offset理应是1030,被压缩之后变成了5，而1030被放在了外层消息上。当消费者消费这个消息集的时候，首先解压整个消息集，找到内层消息中最后一条消息的inner offset， 计算出相应的各个消息的offset,找到相应消息给消费者?

V2版本使用的消息集称之为Record Batch, 不再是之前的Message Set, 它将多个消息(Record)打包存放到单个的RecordBatch中，又通过Varints编码极大地节省了空间。V2版本的消息不仅提供了更多的功能，比如事务，幂等性，某些情况还减少了消息的空间占用，对性能提升很大

• 日志索引

  kafka使用了稀疏索引，每当写入量到一定时(log.index.interval.bytes，默认为4KB)，偏移量索引文件和时间戳索引文件分别增加一个偏移量索引项和时间戳索引项，查询指定偏移量时，使用二分查找法快速定位偏移量的位置，如果指定的偏移量不再索引文件中，会返回小于指定偏移量的最大偏移量。时间戳索引页同理，找到对应的物理文件位置还需要根据偏移量索引进行在此定位。

• 日志删除

  kafka的日志管理器中有一个专门的日志删除任务会周期性地检测和删除segment。周期通过log.retention.check.interval.ms配置，segment的保留策略有三种:

  基于时间：设置一个interval, 根据日志的消息中含有timestamp字段，判断record是否超过该interval，是则删除该record, segment只有部分record符合的时候，会做切分然后再删除

  基于大小: 周期的检查日志的大小寻找需要删除的日志分段。

  基于偏移: 通过DeleteRequest请求设置一个logStartOffset, segment中所有record的offset都小于等于该值时会被删除。

• 日志压缩

  日志的处理除了删除还能设置成压缩，kafka的每条消息都有key这个字段，key在发送的时候可以用来做hash然后将消息发到特定的分区，其在消息压缩时能起到覆盖归类的作用，kafka的日志压缩对应相同Key的不同value值只保存最新的value值

  

深入服务端

• 协议

  kafka中的request和response

  

  字段说明：

  api key:** 比如 PRODUCE FETCH 等分别表示发送消息和拉取消 的请求

  api_ version: api 版本号

  correlation id: 请求id

  client id: 客户端id

---

发消息协议

字段说明：

域(Field)	类型	说明
api_key		0, 表示produce的request
transactional_id	nullable_string	事务id
acks	int16	0(broker收到)， 1(broker写入磁盘)， all(所有副本同步后回ack)
timeout	int32	请求超时时间, (leader-follower同步的时候阻塞需要)
topic_data	array	以topic名字作为分类，topic再以partition作为区分
| | topic	string	topic名
| | data	array	一个array，与主题对应的数据
| | | partition	int32	分区编号
| | | record_set	records	分区对应的数据

这个produceRequest中最重要的是topic_data，可以看到客户端对于消息的处理，会根据topic进行分类，然后根据partition再次进行分类

ProduceResponse协议

域(Field)	类型	说明
throttle_time_ms	int32	服务端配了限流，要求client多少ms后重试
responses	array	对每个消息返回的response
| | topic	string	topic名字
| | partition_responses	array	发送的所有partion的response
| | | partition	int32	分区编号
| | | error_code	int16	错误码
| | | base_offset	int64	消息集的起始偏移量
| | | log_append_time	int64	消息写入broker端的时间
| | | log_start_offset	int64	所在分区的起始偏移量

---

拉消息请求

域(Field)	类型	说明
replica_id	int32	broker: 用作follower指定brokerId发起拉消息请求 普通客户端: -1，不用此值
max_wait_time	int32	与客户端参数fetch.max.wait.ms对应，默认为500
min_bytes	int32	与客户端参数fetch.min.bytes对应，默认1
max_bytes	int32	与客户端参数fetch.max.bytes对应，默认值为52428800
isolation_level	int8	与客户端参数isolation.level对应： read_uncommitted(默认)和read_commited(可选)
session_id	int32	fetch session的id，
epoch	int32
topics	array	要拉取的消息的topic和partition
| | topic		topic名
| | partitions	array	partition名
| | | partition
| | | fetch_offset		指定从那个位置开始读消息
| | | log_start_offset		用于follower发起的request时指明分区的起始偏移量
| | | max_bytes
forgotten_topics_data		指定从fetch session中要去除拉去的 topic--partition消息
| | topic	string	topic名
| | partitions	array	分区名

拉消息的字段主要看topics和forgotten_topics_data这两个，指定了要拉取的消息和不拉取的消息；

由于topics中的字段过多，但消费业务通常都是持续性的，所以引入了session_id， epoch和forgotten_topics_data等， session_id和epoch_id确定了一条拉取链路, 当session建立或变更的时候发送全量的FetchRequest， 当session稳定时发送增量式的FetchRequest请求，里面的topics为空，因为topics域的内容已经被缓存再session两侧，如果需要从当前的fetch session中取消某些分区的拉去订阅，使用forgotten_topis_data字段实现。

拉消息Response

在ResponseBody中可以看到返回的Responses中有对于每一个topic和partition的回复，每一个分区下的消息有对应的partition以及高水位等，最终的消息在record_set中

• 时间轮

  

kafka中使用分层时间轮去处理定时任务，每一层的时间轮上有很多格以1ms作为单位，将定时任务放置在对应的格子内，超过了最大时间格回放到下一层的格子中，下一层的每一个格子的单位是下一层的一圈，整个逻辑上很像时钟的时分秒针的逻辑，通过下层时针的不断转动更新带动上层的转动，不断的更新执行任务；节点的弹出为O(1)，节点的插入为O(1), 节点的删除为O(1) ~ O(n), 通过分层设计，很好的压缩了定时任务从1ms-500ms的稀疏所需要的很多空间。

• 控制器

  在kafka集群中会有一个或多个broker,其中有一个broker会被选举为控制器，负责管理整个集群中所有分区和副本的状态。

  kfka中的控制器的选举工作依赖于Zookeeper，成功竞选为控制器的broker会在zookeeper中创建/controller临时节点，内容如下, 其中broker_id就是被选举的管理者:

  {"version": 1 , ” broker ”: 0 ,”timestamp ”:”1529210278988”}
  

  • 控制器是如何选举的？

  集群中有且只有一个控制器。每个broker启动的时候就回去尝试读取/controller节点中的brokerrid的值，如果读取到的brokerid的值不为-1，则表示已有其它的broker节点成功竞选为控制器，所以当前broker就会放弃竞选；如果zookeeper中不存在/controller节点，或者这个节点中的数据异常，那么就会尝试去创建/controoler节点，只有创建成功的那个broker才会称为控制器。每个broker都会在内存中保存当前控制器的brokerid的值，这个值可以被标志为activeControllerId.

  • 控制器是如何变更?

    zookeeper中还有一个与控制器有关的/controller_epoch节点，这个节点放了一个整形的controller_epoch用来记录控制器发生变更的次数，即当前是第几代控制器，当控制器发生变更时，没选出一个新的控制器就将该字段值+1，每个和控制器交互的请求都会携带controller_epoch这个字段，如果请求的controller_epoch值小于内存中的controller_epoch值，则认为这个请求是向已经过期的控制器所发送的请求，那么这个请求无效.

  • 控制器的负责什么功能?

    监听分区相关的变化: 处理分区重分配的动作,处理ISR集合变更的动作, 处理优先副本的选举动作;

    监听主题相关的变化: 处理主题增减的变化,处理删除主题的动作;

    监听broker的变化: 处理topic的增减的变化

    从zookeeper中读取获取当前所有主题,分区及broker有关的信息并进行相应的管理.

    启动并管理分区状态机和副本状态机.

    更新集群的元数据信息

• kafka的关闭流程

  kafka提供了脚本kafka-server-stop.sh以优雅关闭保证相关数据

  ./kafka-server-stop.sh
  

  但是这个脚本会有bug, 需要对第一行进行修改成如下

  PIDS=${ps ax | grep -i 'kafka' | grep java | grep -v grep | awk '{print $1}'}
  

  脚本中会使用kill -s 去tenimate这个程序,kafka对这个信号做了处理, 会让消息完全同步到磁盘上, 使得下次不需要进行数据恢复操作;同时会对副本进行迁移,减少分区的不可用时间,这个过程叫做ControlledShutdown,见图下:

  

  分区的副本数大于1而且leader副本位于待关闭的broker上,需要实施leader副本的迁移以及新的ISR的变更.这部分由选举器负责;

• 分区leader的选举

  当创建分区或者分区上线时都需要执行一个leader选举的操作, 选举的策略是按照AR集合副本的顺序查找第一个存活的副本,并且这个副本在ISR集合中

• 参数解密

  broker.id: 在集群中,每一个broker都有一个唯一的broker_id用来区分彼此, broker在启动时会在zookeeper中的/brokers/ids路径下创建一个以当前brokerId为名称的虚节点, broker下线时,该虚节点就会被删除; broker.id可以自动生成, 在meta.properties中配置broker.id.generation.enable后即可, 同时可以配置额外的reserverd.broker.max.id配置一个生成的基准线,如默认值为1000,生成的brokerId从1001开始

  bootstrap.servers: 提供一个可以获取元数据的server以获取整个集群的元数据,就可以得到各个broker的地址, 然后与各个broker建立连接.

  

深入客户端

• 分区分配策略

  客户端参数partition.assignment.strategy可以用来设置消费者组中的分区分配策略, 取值范围如下:

  ---

  RangeAssignor(范围):

  Range策略中,将每个topic的区间均匀划分给消费者,有余的话由前面消费者承担

  优点: 相对均匀的讲区间进行划分

  缺点:某些情况下会导致负载失衡;

  

  上图由于区间的数量是3,消费者的数量是2,除不尽导致A要承担多一个分区,如果topic的数量不断的增长,负载失衡会越来越严重

  ---

  RoundRobin(轮转):

  

  轮转分配的方式如上图, 有一个消费者组有A和B两个消费者, 然后我们就可以看到两个topic上的分区以轮转分配的方式分别分配给了A和B;

  优点: 消费者之间的负载很均衡,最多只会相差一个分区

  缺点: 如果一个消费组的消费者订阅的topic不同, 轮转分配只会在订阅的消费者上进行,负载失衡的情况如下

  

  假设消费者c1只订阅了t1, 而消费者c2除了订阅t1还订阅了t2有了4个分区, t1p1无法放到消费者c1上

  ---

  StickyAssignor(黏性):

  由于消费者变更会导致分区重新分配, Range和RoundRobin的重分配上会导致没有必要的区间漂移,为了解决这个问题,在0.11.x版本后搞了个黏性,其首要保证负载均衡,然后保证黏性处理区间, 在再分配过程中尽量保证少的区间调整

  

  上图中解决了轮转分配中跨topic均衡的情况, 在再分配的时候可以看到黏性策略下尽量保证区间不发生变动

  

• 消费者协调器和组协调器

  ---

  多消费者所配置的分配策略并不完全相同,这个分区分配需要协同,这个协同的过程交由消费者协调器和组协调器完成;

  旧版的消费者客户端使用ZooKeeper的监听器维护消费者和偏移进行reblance,每个消费组在ZooKeeper中维护了一个/consumers/<group>/ids路径, 在此路径下使用临时节点记录隶属于此消费组的消费者唯一标志(由消费者创建,consumer.id+主机名+时间戳+UUID, consumer.id是旧版消费者客户端中的配置,相当于新版客户端中的client.id)

  

  与/consumers//ids同级的还有两个节点: owners和offsets, 前者记录了分区和消费者的对应关系,后者记录了消费组在分区中的对应的消费位移.

  每个消费者在启动时都会在/consumers/<group>/ids和/brokers/ids 路径上注册一个监听器(watcher),当/consumers/<group>/ids路径下的子节点发生变化时,表示broker出现了增减, 通过Zookeeper所提供的Watcher, 每个消费者就可以监听消费者组和Kafka集群的变化了.

  每个消费者对Zookeeper的相关路径分别进行监听,当触发再均衡操作时,一个消费组下的所有消费者会同时进行再均衡操作,但存在问题, 消费者之间彼此操作的结果没有进行同步,会导致kafka工作在一个不正确的状态, 同时会引发:

  **羊群效应: **Zookeeper中一个被监听的节点变化,大量的Watcher同指被发送到客户端,导致在通知期间的其它操作延迟;

  脑裂: 消费者进行再均衡操作时候每个消费者都与Zookeeper进行通信以判断消费者或broker变化的情况. 由于Zookeeper本身的特性,可能导致同一时刻各个消费者获取的状态不一致.

  ---

• 再均衡原理和过程

  新版消费者客户端对此进行了重新设计,将全部消费组分成多个自己,每个消费组的子集在服务端对应一个GroupCoordinator(组协处理器)进行管理, 在消费者客户端中ConsumerCoordinator(消费者协处理器)负责与GroupCoordinator进行交互, 再均衡的具体过程如下:

  1.找组协处理器

  消费者向集群中负载最小的节点(leastLoadedNode)发请求找到其对应的GroupCoordinator所在的broker, 集群会根据请求Body的GroupId进行计算查找得到分区编号,然后找到分群leader副本所在的broker节点, 这个就是GroupId所对应的GroupCoordinator节点

  2.加入组

  消费者找到消费组所对应的组协处理器后会向该节点发出JoinGroupRequest请求, 在该请求中会携带GroupId以及Group_Protocols,后者代表该消费者期望对不同的Topic使用的分区策略;组协处理器收到该请求后, 对于分区策略的处理上:生成一个所有消费者都支持的分配策略集合, 对每个消费者在该集合中支持的第一个策略增加一票,票数最多的为最终选举的分配策略; 对于客户端的处理上: 客户端发起JoinGroupRequest时携带了一个memberId字段值为null, coordinator收到后会为此消费者生成一个member_id作为该消费者的标志，这就是旧版的consuner.id，以及新版的client.id配的值; 然后发送一个JoinGroupResponse信令给所有的消费者,

  3.组同步

  消费者组leader收到该信令后得到一个新的分配策略和相关的元信息(消费者,topic等), 通过该策略进行计算一个分区结果, 然后将该结果通过一个SyncGroupRequest请求发送给组协处理器, 同时各个consumer也通过SyncGroupRequest去拉取这个新分配的分区结果.

  4.HeartBeat

  得到一个新的分区结果之后, 消费者使用这个策略就开始正常工作了, 同时向组协处理器发送心跳维持关系,一旦心跳停止就被认为消费者崩溃, 则会重新平衡

  

• __consumer_offsets的剖析

  __consumer_offsets是新版kafka用来存储consumer的消费offset的topic，是kafka唯二两个特殊的topic，还有一个叫做transaction(用来协助处理kafka事务的)，__consumer_offsets中通过record的方式存储consumer的偏移，

---

• 消息传输保障

  在消息传输保障上,由于生产者向kafka发消息由于多副本机制,消息肯定不会丢失,但是由于网络原因(ack丢失)而导致重新发送, 所以生产者是at least once; 在消费时结果取决于处理消息的方式,如果是拉了消息后处理完了再commit offset, 有可能因宕机没来得及commit 此时是at least once, 但如果是先commit后做处理消息, commit之后宕机有可能丢失消息,此时是at most once,为了实现exactly once, 引入了幂等和事务

• 幂等
  无论多少次调用结果所产生的影响都是一样的,要开启幂等需要将生产者客户端参数enable.idempotence设置成true

  幂等是怎么实现的?

  通过producer id 和 sequence实现的; kafka引入了 producer id和序列号这两个概念，分别对应v2格式中的RecordBatch的producer id和first sequence这两个字段，每个新的生产者实例在初始化得时候都会分配一个PID，对于每一PID，消息发送到每一个分区都有相应得序列号，从0开始单调递增。生产者每发送一条消息就会将<PID, 分区>对应得值加1，broker端维护这些值，消息只有严格得按照序列递增才会被接收，序列号小于等于当前值得时候为重复会被kafka丢弃，大于得时候说明消息丢失了，broker会告诉生产者

• 事务

  kafka中的事务于mysql中的acid并不是一个东西，它是为了解决两个问题，1.kafka的幂等性只是针对每一个分区而言的，而事务可以提供多个分区写入操作的原子性。2.为了解决read-process-write问题，从一个旧的topic中读取数据，然后处理，将处理结果生产到一个新的topic中去

  kafka事务的开启:

  1.将transaction.id参数设置为非空同时将enable.idlempotence设置为true

  kafka的事务的隔离级别？

  在客户端配置，分别有read_commited和read_uncommited 两种，前者不能读到生产者未提交的数据而后者能

  客户端读取到read_uncommited消息时的seq变化？

  由于消费不到offset_consumer暂时不得而知

  事务是如何实现的?

  利用transactionid， produce id和transaction epoch实现的

  在使用事务时，应用程序必须手动显示提供transactionId, 程序会使用它在kafka中注册，在应用宕机重启后kafka能够识别出该应用从而结束未结束的的事务；一个客户端实例起来后会使用该transactionId向kafka注册后会获得一个producer id(用来提供生产幂等性)

  同时应用使用transaction id注册时候会分配一个producer epoch的东西，使用同样的transaction id去kafka注册的时候epoch的值会加一，分配给客户端，使用同样的epoch id但是携带旧的epoch的有关请求就无效了，通过epoch解决了一个问题，假设应用发了一个与该transaction id有关的操作出去还未到kafka, 此时应用宕机重启后开启一个了事务，这个在网络上卡了很久的tcp包到了kafka，这就影响了新事务的正常运行，增加epoch避免了这个问题

  为了更清楚的理解kafka中事务是如何实现的，需要贴一段程序来全局认知

      public static void main(String[] args) {
          Consumer<String, String> consumer = createConsumer();
          Producer<String, String> producer = createProduceer();
  
          // 初始化事务
          producer.initTransactions();
  
          while(true) {
              try {
                  // 1. 开启事务
                  producer.beginTransaction();
                  // 2. 定义Map结构，用于保存分区对应的offset
                  Map<TopicPartition, OffsetAndMetadata> offsetCommits = new HashMap<>();
                  // 2. 拉取消息
                  ConsumerRecords<String, String> records = consumer.poll(2000);
                  for (ConsumerRecord<String, String> record : records) {
                      // 3. 保存偏移量
                      offsetCommits.put(new TopicPartition(record.topic(), record.partition()),
                              new OffsetAndMetadata(record.offset() + 1));
                      // 4. 进行转换处理
                      String[] fields = record.value().split(",");
                      fields[1] = fields[1].equalsIgnoreCase("1") ? "男":"女";
                      String message = fields[0] + "," + fields[1] + "," + fields[2];
                      // 5. 生产消息到dwd_user
                      producer.send(new ProducerRecord<>("dwd_user", message));
                  }
                  // 6. 提交偏移量到事务
                  producer.sendOffsetsToTransaction(offsetCommits, "ods_user");
                  // 7. 提交事务
                  producer.commitTransaction();
              } catch (Exception e) {
                  // 8. 放弃事务
                  producer.abortTransaction();
              }
          }
      }
  

  其中最主要的1.开启事务，5.提交偏移量，7.提交事务，接下来简述具体过程

  

  1.查找TransactionCoordinator

  应用发送TransactionId给任意Broker, Broker收到之后使用Utils.abs(transactionalId.hashCode)%transactionTopicPartitionCount查找被分配到的分区，然后根据分区找到leader节点，这个节点就是这个transactionId所对应的TransactionCoordinator节点。

  2.获取Pid

  找到TransactionCoordinator节点之后，该节点会给该TransactionId分配一个item用来维护这个使用该TransactionId的应用事务状态，该item内容如下图，正如我们所想，以transactional_id为key还额外携带了一个version是用来维护版本；

  在value中特别需要注意的是transaction_status，它标记着transaction_id的状态，比如说Empty(0)未开始, CompleteCommit(4)已提交，Dead(6)死亡异常等; 还有一个要注意的是transaction_partition，记录了事务操作的topic和partition

  

  如果transactional_id对应的item已存在，第二次构造该item会对其重写，对其中的producer_epoch+1，使得网络上残存的关于该事务的包无效，根据transaction_status进行状态重置(Commit或Abort)

  3.开启事务

  在上述代码中的1.开启事务被运行后，生产者本地会标记开启一个新事务，但是要到发送第一条消息之后，TransactionCoordinator才会认为已开启，进行item更新

  4.发消息

  在代码的5. 生产消息到dwd_user中生产者给一个新的分区发送数据前，会先向TransactionCoordinator发送AddPartitionsToTxnRequest请求，这个请求会把topic和partition_id写入到item的transaction_partitions中，然后才把消息发送到topic的对应分区上

  5.提交偏移量

  在6. 提交偏移量到事务中将消费者所消费的offset进行提交，TransactionCoordintor收到该请求之后提取其中的groupId使用Math.abs(GroupId.hashCode)%Partition_Count计算其__consumer_offsets所在的分区，把这些信息也写到item中的transaction_partitions中；

  然后客户端会发送请求让Coordinator将offset写入到__consumer_offsets中

  6.提交/中止事务

  在7.提交事务中，应用会发送请求到Coordinator让其Commit事务，Coordinator做了一个两阶段提交：1.首先将该item的status改成PREPARE_COMMIT、2.然后发送请求WriteTxnMarkersRequest将Commit信息写入用户的普通Topic和__consumer_offsets、3.成功之后将Item改成COMPLETE_COMMIT；

  其中WriteTxnMarkersRequest会向item中记录的各个分区发送，收到了该请求之后partition会增加一个control record标志着一个事务的结束，它的attributes中的第6位被标志成1，标志这是一条控制消息而不是普通的数据消息，所有属于该事务的record的attributes第5位都被标志成1，将这些消息在划分在事务中，最后的control record的type标志着这个事务是commit(1)或者abort(0)；那么别的客户端设置了read_commited的时候就需要读取到最后一条control record才能判断这些消息是否commit才进行处理(但话说回来多生产者下的话不就无法保证有序消费，毕竟事务中可能夹杂着其余生产者的消息，还是说也保证了但是其它生产者的消息也被整个事务阻塞了，只有结束record后，才会从事务的第一条开始有序处理)

可靠性探究

• 名词释意

  AR： 分区中所有的副本统称为AR
  ISR: 能够与leader保持同步状态的副本的集合

  LEO: 每个分区中最后一条消息的下一个位置

  HW: 低水位，ISR中最小的LEO即为HW

• 失效副本

  无法在规定的时间或者规定的条数之内和leader保持数据同步的副本称为失效副本，也就是ISR之外的副本，可以用该命令进行查看

  ./bin/kafka-topics.sh --bootstrap-server localhost:9092
  --describe
  --topic topic-create
  --under-replicated-partitions
  

  broker端有一个参数replica.lag.time.max.ms, 默认为10000，当follower副本之后leader副本的时间超过该参数时判定为同步失败，当follower副本将leader副本的(LEO)之前的日志都同步后，此时就认为follower副本已经追赶上leader副本了，会记录下该时间为LastCaughtUpTimeMs, 然后kafka的副本管理器会启动一个定时任务去判断 LastCaughtUpTimeMs与当前的时差是否超过了上面设置的超时时间

• ISR是如何维护管理的？

  kafka启动了两个定时任务，1.isr-expiration. 2.isr-change-propation

  isr-expiration周期检测是否要将不符合ISR集合的副本剔除出ISR，使用的方法便是利用上面所说的LastCaughtUpTimeMs, 当检测到有失效副本，就会收缩ISR集合。ISR集合变更后的结果会被记录到ZooKeeper对应的/brokers/tpics/<topic>/patition/<partition>/state

  {"controller_epoch":26, "leader"s:0,"version":1,"leader_epoch":2, isr:{0,1}}
  

  同时还会将该记录缓存到isrChangeSet中，isr-change-propation会周期性检查isrChangeSet，将其中的有效数据放入到zookeeper中的/isr_change_notification/isr_change....节点中，kafka控制器为notification添加了一个Watcher，当这个节点中有子节点发生变化的时候触发watcher的回调去告诉相关broker更新源数据信息，然后删除/isr_change_notification下的有关节点。

• LEO与HW是如何变化

  1.leader副本的LEO增加到5，此时所有副本的HW均为0。

  2.follower向leader拉取消息发送FetchRequest请求，其中带有自己的LEO请求，leader收到后通过请求中的LEO，是否比ISR集合的HW小，如果是则更新，返回相应的消息，以及自身的HW信息。

  3.follower收到后更新自己的HW，为当前自己的LEO与leader发送过来的HW的最小值。

• leader切换的同步过程是怎么样的？

  0.11.0.0版本之前使用的是基于HW的同步机制，这种同步方式会导致数据丢失，其丢失场景如下：

  

  A作为follower从leader B中拉取写入消息 m2 之后(LEO 更新)需要再一轮的 FetchRequest/ FetchResponse 才能更新自身的HW为2 .如果在这个时A宕机了，那么在A重启之后，首先会根据之前 HW 位置（这个值会存入本地的复制点文件 replication-offset-checkpoint ）进行日志截断这样便会将 m2 这条消息删除 ，此时A剩下m1一条消息，此时B再宕机，那么A就会被选举为新的leader，B恢复后称为了follower，副本的Hw不能比leader副本的HW高，否则就会做日志截断，所以m2消息将会被删除

  为了解决上面的问题，0.11之后引入了leader epoch, 其代表leader的纪元信息，每当leader变更一次,它的值就会加1，引入了leader epoch之后, A在复制写完消息后还没来得及更新HW就宕机了，此时重启后不是首先忙着截断日志而是先发送OffsetsForLeaderEpochRequest请求给B，B作为leader收到请求之后会返回当前的LEO做同步；

  

  然而如果碰巧此时A(leader),B(follower)都挂了，而B被选举为leader并且接收了额外的消息(m3)，B成为了follower，B重启后向leader请求获得HW，此时发现A和B的leader epoch不一样，A会将leader epoch 为B的leader epoch+1的第一条消息offset发送给B，然后B就会做日志截断，把A的消息同步过来

• kafka的可靠性

  kafka和redis不一样不支持读写分离是因为kafka的消息要做存在硬盘要经历，主从复制需要经历网络->主节点内存->主节点磁盘->网络->从节点内从->从节点磁盘，这两个操作非常耗时；所以kafka搞了个分区去做负载均衡，当然redis集群也有hash slot这种东西其实概念是一样的。主写从读的方式能通过读写分离的方式实现，当时主写主读的负载均衡就只能通过分区实现。

  kafka的可靠性基于ISR，而redis的quorum模型保证可靠性，ISR的优势在于降低了副本的数量，因为quorum模型的必须过半选举的特性为了容忍f个实例的失败，必须提供2f+1个实例, 否则无法选择新的leader,而ISR一共只需要提供f+1个副本（有人能出来当老大即可）,但是这种方式有一个弊端，如果将ack设置为1(等待follower将消息存储完毕之后才给客户端发送response)，那么就需要等待所有的副本都存完消息；而quorum不用，它只需要过半存盘即可，这种特性使得它的系统延迟取决于最快的几个节点，而kafka没法办做到，所以把这个问题抛给了客户端通过ack去自己做选择。

  越多的副本虽然能提供更高的可靠性，但为之而来的网络IO等开销也会随之增大，副本数为3已经可以满足绝大多数的场景，对于高可靠性要求的场景，这个数值为5

总结

kafka将逻辑结构为broker-topic-partition, partition作为是逻辑最细的单位，其下面存储着records，每一条record都有offset用于消费和管理，将records进行segment能加速查找以及方便管理(删除或压缩record)，record的存储方式上是一个.log文件和.index文件, 通过offset和.index文件可以快速查找到record所在的位置；

kafka使用了partition多备的方式提供可靠性，follower partition作为备点需要和leader partition做数据同步，能在规定时间内完成数据同步的副本被放入一个叫做ISR的集合，当leader宕机后, zookeeper上的维护leader节点状态发生了变更，所有的follower都watch了该节点都收到了事件，属于ISR内得follower会去抢占创建zookeeper节点，最后按照先到先得方式成为新leader完成选举，由于follower的hw更新在下一个fetchResponse后才能得到更新，会导致leader变更后的数据同步存在导致数据丢失以及ABA以及HW不正确的问题，为此搞了一个leader epoch来防止这些问题；而不能在规定时间内完成数据同步得副本无法放入ISR中，它们统称为失效副本

生产者作为数据源给kafka发送数据，发送数据的时候有key-value，key用来做hash到不同的分区以实现均衡，value就是我们说的record，发送的过程中并不是一条一条发送的，客户端会将其进行打包成为一个batch集合进行发送，同时可以通过生产者客户端的ack参数配置消息发送可靠性，分别有 发出去就成功->broker刷盘才算成功->follower都同步完才算成功 需要结合业务进行选择

消费者会定时的从服务端轮询是否有新的消息，消费是以消费组的方式对topic进行消费的，topic会通过策略将partition进行划分给消费组内的消费者去同步消费，分配策略有range、roundrobin、sticky(推荐)，它们在消费者客户端进行配置最后由消费者协调器统一协调选取出来，当消费组成员发生数量变化的时候，会进行reblance(再均衡), 也就是对分区会根据策略重新分配；对相关的record进行消费后，客户端会提交消费偏移，kafka维护了一个名consumer_offset的topic保存这些偏移，由于这个消费后的offset提交是客户端自己做的，需要考虑到多线程以及一致性；

为了解决read-process-write是一个流式处理中常面临的事务场景，kafka引入了producer id 和seq 实现了producer的exactly once，同时利用producer_epoch、transaction id和topictransaction_states实现二次提交最终去解决