Kafka核心架构图

1、主线程创建消息对象。2、对消息进行序列话，使其能够在网络上传输3、经过分区器，确认消息发送的目标分区4、由sender县城发送消息。根据不同的acks策略，监听Kafka服务端返回的消息发送结果，当发送错误时，根据重试配置进行重试

TopicB#Partition1#repition1(follower)

Segment 2

。消费者在消费topic中消息时，只与分区中leader的分区副本进行消息poll。。每个消费者从属于一个消费者组，一个消费者组中可以有多个消费者，一个消费者可以订阅多个topic。一个topic中的分区只可以被一个消费组中的消费者消费，一个消费者可以消费一个topic中的多个分区。。当一个消费者组中的消费者数量大于一个topic中分区数量时，存在消费者空闲的情况

topic名分区号-消费者组名消费者名

Topic _consumer_offsets

TAP0-G2C1

zookeeper2

Broker3

TopicB#Partition2#repition3(follow)

消费者注册与管理：

Kafka 分区leader选举过程

Interceptors过滤器

consumer1

consumer2

Segment n

分区物理存储

TAP0-G1C1

TopicB#Partition1#repition2(leader)

消费者poll消息

Kafka controller

TopicA#Partition1#repition1(follower)

Serializer序列化器

TBP2-G2C1

发送消息时的分区器采用的分区策略：1、发送消息时指定分区号，则发往指定topic分区。2、发送消息时设置了key，则对key进行hash在对该      topic的分区数进行取模，则得到分区需要的分区号，即hashcode(key)%partitionNum。3、发送消息时既没指定分区号，也没指定key，则使用轮询算法（round robin）获取分区号。

1、每个Kafka集群有且仅有一个broker选为controller，主要负责分区leader的选举2、第一个启动的broker向zookeeper里创建/controller让自己成为controller，其他启动的broker也尝试创建/controller，但是会被告知/controller已存在。3、其他broker会意识到已经选好了controller broker，则自己在zookeeper上创建watch对象，来接受/controller的变更消息。4、如果controller broker断开链接，则其他broker会收到watch的控制器节点消失的消息，其他节点会尝试将自己注册为/controller节点，先到先得5、如果controller发现一个broker消失，那些在该broker上存储的leader分区也一同消失，controller则需要对生育follower分区进行遍历，选择一个新的leader分区，并通知其他follower分区

Consumer

zookeeper cluster

1、分区副本在概念上有AR与ISR之分。2、AR表示分区向zookeeper注册过的所有副本序列（assigned replicas）3、ISR：和leader副本保持同步的follower副本集合序列。如果follower副本长时间没有向leader保持通信或者没有同步数据，则该follower会被提出ISR。时间阈值由replicas.lag.time.max.ms控制，默认30s，leader发生故障后会从ISR中选举新的leader4、OSR：与leader副本同步时，出现超时的follower副本集合5、AR =ISR+OSR6、分区的ISR信息可以从zookeeper里查看，路径：/brokers/topics/{topin_name}/{}partitions/{partition_num}/state

producer1

Producer

TAP1-G2C2

TAP2-G2C1

TopicA#Partition1#repition3(follower)

acks

......

TopicB#Partition0#repition3(leader)

TopicB#Partition2#repition2(follow)

TAP2-G1C3

Partition1#repition1(follower)

Broker2

Segment 1

xxx.indexxxx.timeindexxxx.logxxx.txnindex

1、消费者不断向kafka集群poll消息2、Kafka向消费者返回一批次消息3、消息经过反序列化器，进行反序列化解析4、消息进过过滤器5、消费者消费消息6、一批次消息消费完后，向系统内置topic：_consumer_offsets提交当前批次最大的offsets

分区leader副本（红色边框）

Partitioner分区器

consumer3

Kafka 分区副本基本信息

Kafka cluster

TopicB#Partition0#repition1(follower)

TopicB#Partition0#repition2(follower)

Kafka topic分区文件存储系统

TopicA

发送消息步骤（生产者只与leader分区副本进行消息发送）：

TopicA#Partition1#repition2(leader）

producer3

TopicA#Partition0#repition3(leader)

zookeeper0

commit offset

send ProduceRecord

分区文件系统

Broker1

消费者消费过程：

图例

TBP2-G2C2

TopicA#Partition0#repition1(leader)

TopicA#Partition2#repition3(leader)

TopicB

Kafka生产者消息 ProduceRecordkey value

1、一个topic消息按分区进行存储2、存储位置为Kafka broker中server.properties中log.dirs配置的值3、分区文件目录命名为：topic名+分区号4、分区采用段（segment）进行存储。5、一个分区目录下的文件有：xxx.log、xxx.index、xxx.txnindex等6、xxx.log为消息被序列化后存储的实际消息文件，xxx.index为消息的索引文件，采用稀疏索引。“xxx”表示该文件位于该分区的下的起始offset7、消息文件会在一定时间被清除，可配置清除时间和策略，可配置为压缩

TopicB#Partition1#repition3(follower)

group-2

TopicB#Partition2#repition1(leader)

分区follower副本（黑色边框）

TopicA#Partition2#repition2(follow)

group-1

TAP1-G1C2

producer2

TBP0-G2C1

zookeeper1

Sender线程

消费者消费原则：

1、每一个broker中都有一个组件coordinator负责管理消费者与消费者组。2、消费者创建时首先向消费者组中注册。3、消费者组需要选取自己的coordinator。根据hashcode(groupId)%50(_cunsumer_offsets分区数)结果，选择该结果的所在的_consumer_offsets的分区躲在的broker的coordinator作为该消费者组的管理者。4、消费组中的所有消费者向coordinator中请求加入到该组5、coordinator在收到消费者加入组请求后，随机选取一位消费者作为该组中的leader。6、coordinator向消费者leader发送topic详情。7、消费组leader根据topic详情指定消费方法，哪个消费者消费哪个分区，并发送给coordinator。8、coordinator在收到方案后，向组中所有消费者发送该方案。9、消费组需要与coordinator保持心跳通信，确保消费者存活，如果超过时间未通信，coordinator将该消费者从消费者组中提出，并触发再平衡。

Partition49#repition1(leader)

TopicA#Partition2#repition1(follow)

TopicA#Partition0#repition2(leader)

Partition1#repition0(follower)