Kafka源码解析与实战

Kafka的架构

包括Kafka的基本组成，Kafka的拓扑结构以及Kafka的内部通信协议。Kafka内部的通信协议是建立在Kafka的拓扑结构之上，而Kafka的拓扑结构是由Kafka的基本模块所组成的。
AK RELEASE 2.5.0
APRIL 15, 2020

Kafka的基本组成

Kafka集群中生产者将消息发送给以Topic命名的消息队列Queue中，消费者订阅发往以某个Topic命名的消息队列Queue中的消息。其中Kafka集群由若干个Broker组成，Topic由若干个Partition组成，每个Partition里的消息通过Offset来获取。
基本组成包括：

Broker：一台Kafka服务器就是一个Broker，一个集群由多个Broker组成，一个Broker可以容纳多个Topic，Broker和Broker之间没有Master和Standby的概念，他们之间地位是平等的
Topic：每条发送到Kafka集群的消息都属于某个主题，这个主题就称为Topic。物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存在一个或多个Broker上，但是用户只需指定消息主题Topic即可生产或消费数据而不需要关心数据存放在何处。
Partition：为了实现可扩展性，一个非常大的Topic可以被分为多个Partition，从而分布到多台Broker上。Partition中的每条消息都会被分配一个自增Id（Offset）。Kafka只保证按照一个Partition中的顺序将消息发送给消费者，但是不保证单个Topic中多个Partition之间的顺序。
Offset：消息在Topic的Partition中的位置，同一个Partition中的消息随着消息的写入其对应的Offset也自增。
Replica：副本，Topic的Partition有N个副本，N为副本因子。其中一个Replica为Leader，其他都为Follower，Leader处理Partition的所有读写请求，Follower定期同步Leader上的数据。
Message：消息是通信的基本单位。每个Producer可以向一个Topic发布消息
Producer：消息生产者，将消息发布到指定的Topic中，也能够决定消息所属的Partition：比如基于Round-Robin或者Hash算法
Consumer：消息消费者，向指定的Topic获取消息，根据指定Topic的分区索引及其对应分区上的消息偏移量来获取消息
Consumer Group：消费者组，每个消费者都属于一个组。当消费者具有相同组时，消息会在消费者之间负载均衡。一个Partition的消息只会被相同消费者组中的某个消费者消费。不同消费者组是相互独立的。
Zookeeper：存放Kafka集群相关元数据的组件。Zookeeper集群中保存了Topic的状态信息，例如分区个数、分区组成、分区的分布情况等；保存Broker的状态信息；保存消费者的消费信息等。通过这些信息，Kafka很好地将消息生产、消息存储、消息消费的过程结合起来。

Kafka的拓扑结构

一个典型的Kafka集群中包含若干个Producer（可以是某个模块下发的Command，或者是Web前端产生的PageView，或者是服务器日志，系统CPU、Memory等），若干个Broker（Kafka集群支持水平扩展，一般Broker数量越多，整个Kafka集群的吞吐率也就越高），若干个Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置。Producer使用Push模式将消息发布到Broker上，Consumer使用Pull模式从Broker上订阅并消费消息。

简单的消息发送流程如下：

Producer根据指定的路由方法，将消息Push到Topic的某个Partition里
Kafka集群接收到Producer发过来的消息并持久化到硬盘，并保留消息指定时长（可配置），不关注消息是否被消费。
Consumer从Kafka集群Pull数据，并控制获取消息的Offset

Kafka内部的通信协议

Kafka内部各个Broker之间的角色并不是完全相等的，Broker内部负责管理分区和副本状态以及异常情况下分区的重新分片等这些功能的模块称为KafkaController。每个Kafka集群中有且只有一个Leader状态的KafkaController，当其出现异常时，其余Standby状态的KafkaController会通过Zookeeper选举出有一个Leader状态的KafkaController。

维度一：通信协议详情

ProducerRequest：生产者发送消息的请求，生产者将消息发送至Kafka集群中的某个Broker，Broker接收到此请求后持久化此消息并更新相关元数据信息。
ProducerRequest.requiredAcks的取值为0时，生产者不关心Broker Server端持久化执行结果，但是高级消费者发送的提交偏移量的请求还是需要返回具体执行结果。为1则生产者消费者都需要将Broker Server端持久化的执行结果返回客户端。为-1时，不会立刻返回Broker Server端消息持久化的结果，而是需要等待Partition的ISR列表中的Replica完成数据同步，并且ISR列表的个数大于min.insync.replicas时才会将响应返回给对应客户端。这里采用的是称为Purgatory的策略。Broker Server上对应的Partition的HighWatermark发生改变才触发检查。
TopicMetadataRequest：获取Topic元数据信息的请求，无论是生产者还是消费者都需要通过此请求来获取感兴趣的Topic的元数据。
FetchRequest：消费者获取Topic的某个分区的消息的请求。分区状态为Follower的副本也需要利用此请求去同步分区状态为Leader的对应副本数据。
OffsetRequest：消费者发送至Kafka集群来获取感兴趣Topic的分区偏移量的请求，通过此请求可以获知当前Topic所有分区在不同时间段的偏移量详情。
OffsetCommitRequest：消费者提交Topic被消费的分区偏移量信息至Broker，Broker接收到此请求后持久化相关偏移量信息。
OffsetFetchRequest：消费者发送获取提交至Kafka集群的相关Topic被消费详细信息，和OffsetCommitRequest相互对应。
LeaderAndIsrRequest：当Topic的某个分区状态发送变化时，处于Leader状态的KafkaController发送至相关Broker通知其做出相应处理。
当某个Replica称为Leader：暂停Fetch→添加进Assigned Replica列表→添加进In-Sync Replica列表→删除已经不存在的Assigned Replica→初始化Leader Replica的HighWatermark
当某个Replica成为Follower：暂停旧的Fetch线程→截断数据至HighWatermark以下→开启新的Fetch线程→添加进Assigned Replica列表→删除已经不存在的Assigned Replica
StopReplicaRequest：当Topic的某个分区被删除或者下线的时候，处于Leader状态KafkaController发送至相关Broker通知其做出相应处理。
UpdateMetadataRequest：当Topic的元数据信息发生变化时，处于Leader状态的KafkaController发送至相关Broker通知其做出相应处理。
BrokerControllerShutdownRequest：当Broker正常下线时，发送此请求到处于Leader状态的KafkaController。
ConsumerMetadataRequest：获取保存特定Consumer Group消费详情的分区信息

维度二：通信协议交互

Producer和Kafka集群：Producer需要利用ProducerRequest和TopicMetadataRequest来完成Topic元数据的查询、消息的发送。
Consumer和Kafka集群：Consumer需要利用TopicMetadataRequest请求、FetchRequest请求、OffsetRequest、OffsetCommitRequest、OffsetFetchRequest、ConsumerMetadataRequest来完成Topic元数据的查询、消息的订阅、历史偏移量的查询、偏移量的提交、当前偏移量的查询。
KafkaController状态为Leader的Broker和KafkaController状态为Standby的Broker：Leader需要用LeaderAndIsrRequest、StopReplicaRequest、UpdateMetadataRequest来完成对Topic的管理。Standby需要利用BrokerControllerShutdownRequest来通知Leader自己的下线动作。
Broker和Broker之间：Broker相互之间需要利用FetchRequest请求来同步Topic分区的副本数据，这样才能使Topic分区各副本数据保持一致。

Broker概述

Broker内部存在的功能模块包括SocketServer、KafkaRequestHandlerPool、LogManager、ReplicaManager、OffsetManager、KafkaScheduler、KafkaApis、KafkaHealthcheck和TopicConfigManager九大基本模块以及KafkaController集群控制管理模块。

SocketServer：首先开启一个Acceptor线程，新的Socket连接成功建立时会将对应的SocketChannel以轮询方式转发给N个Processor线程中的某一个，由其处理接下来SocketChannel的请求，将请求放置在RequestChannel中的请求队列；当Processor线程监听到SocketChannel请求的响应时，会将响应从RequestChannel中的响应队列中取出来并发给客户端
KafkaRequestHandlerPool：真正处理Socket请求的线程池，其个数默认为8个，由参数num.io.threads决定。该线程池里面的线程KafkaRequestHandler从RequestChannel的请求队列中获取Socket请求，然后调用KafkaApis完成真正业务逻辑最后将响应写回至RequestChannel中的响应队列，并交给SocketServer中对应的Processer线程发给客户端。
LogManager：Kafka的日志管理模块，主要提供针删除任何过期数据和冗余数据，刷新脏数据，对日志文件进行Checkpoint以及日志合并的功能。
负责提供Broker Server上Topic的分区数据读取和写入功能，负责读取和写入位于Broker Server上的所有分区副本数据。如果Partition有多个Replica，则每个Broker Server不会存在相同Partition的Replica，如果存在一旦遇到Broker Server下线会丢失Partition的多份副本可用性降低。
LogManager中包含多个TopicAndPartition，每个TopicAndPartition对应一个Log，每个Log中包含多个LogSegment（每个LogSegment文件包括一个日志数据文件【.log】和两个索引文件（偏移量索引文件【.index】和消息时间戳索引文件【.timeindex】））。LogSegment的结构中log代表消息集合，每条消息都有一个Offset，这是针对Partition中的偏移量；index代表的是消息的索引信息，以KV对的形式记录，其中K为消息在log中的相对偏移量，V为消息在log中的绝对位置；baseOffset代表的是该LogSegment日志段的起始偏移量；indexIntervalByte代表的索引粒度，即写入多少字节之后生成一条索引。OffsetIndex不会保存每条消息的索引，因此其索引文件是一个稀疏索引文件（稀疏索引：索引项中只对应主文件中的部分记录，即不会给每条记录建立索引）。

后台还会维护一个日志合并线程，Kafka发送消息的时候需要携带3个参数（Topic，Key，Message），针对相同的Key值不同的Message只保留最后一个Key值对应的消息内容。

ReplicaManager：Kafka副本管理模块。主要提供针对Topic分区副本数据的管理功能，包括有关副本的Leader和ISR的状态变化、副本的删除、副本的监测等。ISR全称是In-Syn Replicas，处于同步状态的副本。AR全称是Assign Replicas的缩写，代表分配给Partition的副本。
主要利用ReplicaFetcherThread（副本数据拉取线程）和High Watermark Mechanism（高水位线机制）来实现数据的同步管理。单个ReplicaFetcherThread线程负责某个Broker Server上部分TopicAndPartition的Replica数据同步。将拉取的消息写入log，更新当前Replica的HighWatermark，代表的是ISR中所有replicas的last commited message的最小起始偏移量。当某个Broker Server上被分配到Replica的时候会进入becomeLeaderOrFollower处理流程；当Replica被删除为进入stopReplicas处理流程；当Follower状态的Replica长时间没有同步Leader状态的Replica的时候会进入maybeShrinkIsr处理流程。
OffsetManager：Kafka的偏移量管理模块，主要提供针对偏移量的保存读取的功能，两种方式保存：一种是把偏移量保存到Zookeeper上，另一种是Kafka，把偏移量提交至Kafka内部Topic为"__consumer_offsets"的日志里面，主要由offsets.storage参数决定。默认为Zookeeper。
KafkaScheduler：Kafka的后台任务调度资源池。提供后台定时任务的调度，主要为LogManager、ReplicaManager、OffsetManager提供调度服务。
KafkaApis：Kafka的业务逻辑实现层，根据不同的Request执行不同的操作，其中利用LogManager、ReplicaManager、OffsetManager来完成内部处理。
KafkaHealthcheck：Broker Server在./brokers/ids上注册自己的ID，当Broker在线的时候，则对应的ID存在；当离线时对应ID不存在，从而达到集群状态监测的目的。
TopicConfigManager：在/config/changes上注册自己的回调函数来监测Topic配置信息的变化
KafkaController：Kafka集群控制管理模块。由于Zookeeper上保存了Kafka机器的元数据信息，因为KafkaController通过在不同目录注册不同的回调函数来达到监测集群状态的目的，以及响应集群状态的变化：
1. /controller目录保存了Kafka集群中状态为Leader的KafkaController标识，通过监测这个目录的变化可以即使响应KafkaController状态的切换
2. /admin/reassign_partitions目录保存了Topic重分区的信息，通过监测这个目录的变化可以及时响应Topic分区变化的请求
3. /admin/preferred_replica_election目录保存了Topic分区副本的信息，通过监测这个目录的变化可以及时响应Topic分区副本变化的请求。
4. /brokers/topics目录保存了Topic的信息，通过监测这个目录的变化可以及时响应Topic创建和删除的请求。
5. /brokers/ids目录保存了Broker的状态，通过监测这个目录的变化可以及时响应Broker的上下线情况。

Broker的控制管理模块

每个Broker内部都会存在一个KafkaController模块，但是有且只有一个Broker内部的KafkaController模块对外提供控制管理Kafka集群的功能，例如负责Topic的创建、分区的重分配以及分区副本Leader的重新选举等。

KafkaController的选举策略

Leader和Follower的选举是基于Zookeeper实现的，尝试在Zookeeper的相同路径上创建瞬时节点（Ephemeral Node），只有一个KafkaController会创建成功。其中负责状态管理的类为ZookeeperLeaderElector，字面意思上就可以看出是基于Zookeeper的Leader选举权。其中包含了controllerContext当前Topic的元数据信息以及集群的元数据信息等；electionPath为多个KafkaController竞争写的路径，其值为/controller；onBecomingLeader为状态转化成Leader时候的回调函数；onResigningAsLeader为状态转化位Follower时候的回调函数；brokerId为当前Broker Server的Id。ZookeeperLeaderElector启动后负责观察数据节点状态，瞬时节点消失触发再次选举，尝试写入的节点内容就是brokerId。

KafkaController的初始化

当选举为Leader时分为下面几步：

初始化Kafka集群内部的时钟，存放在Zookeeper的/controller_epoch，Broker Server用这个值区分请求的时效性
注册各种监听函数，针对Zookeeper不同目录下Kafka存储的不同元数据进行监听。
通过initializeControllerContext()、replicaStateMachine.startup()和partitionStateMachine.startup()初始化Kafka集群内部元数据信息。建立和集群内其他状态为Follower的KafkaController的通信链路，处理集群启动前没有及时处理的用户请求，此时可能会变更Kafka集群内部的元数据信息，最后通过sendUpdateMetadataRequest()将Kafka集群内部的元数据信息同步给其他状态为Follower的KafkaController。
根据auto.leader.rebalance.enable配置项按需启动Kafka集群内部的负载均衡线程，默认开启
根据delete.topic.enable配置项按需启动Kafka集群内部的Topic删除线程，默认关闭

选举为Follower的时候分为下面几步，正好与Controller相反：

取消Zookeeper路径上的监听函数
根据delete.topic.enable配置项按需启动Kafka集群内部的Topic删除线程，默认关闭
关闭Kafka集群内部的负载均衡线程
断开和集群内其他状态为Follower的KafkaController的通信线路
重置集群内部时钟

Topic的分区状态转化机制

Topic的分区状态维护是由PartitionStateMachine模块负责的，通过在/brokers/topics 和 /admin/delete_topics目录上注册不同的监听函数，监听Topic的创建和删除事件触发Topic分区状态的转换。
PartitionStateMachine中分区状态由PartitionState用一个字节表示不同状态，分为四种：

NonExistentPartition：代表分区从来没有被创建或者被创建之后又被删除呃状态
NewPartition：分区刚创建包含AR，但是此时Leader或ISR还没有创建，处于非活动状态无法接收数据
OnlinePartition：分区Leader已经被选举，产生来对应的ISR，处于活动状态可以接收数据
OfflinePartition：代表分区Leader由于某种原因下线时导致分区暂时不可用

每个状态都是由一个合理的前置状态转换而来。

Topic分区的领导者副本选举策略

Topic分区的Leader Replica在不同场景下的选举策略是不一样的，不同选举策略都基础PartitionLeaderSelector。其根据Topic、Partition、当前Leader、当前的ISR选举出新的Leader，新的ISR和新的AR（在线状态），共有5种不同的策略：

NoOpLeaderSelector：默认的选举策略
ReassignedPartitionLeaderSelector：当分区AR重新分配时使用的策略
PreferredReplicaPartitionLeaderSelector：集群内部自动平衡负载或者用户触发手动平衡负载时使用的策略
随着Topic的新建删除以及Broker Server的上下线，原本Topic分区的Leader Replica在集群中的分布越来越不均匀。 auto.leader.rebalance.enable为true，则会自动触发分区的Leader Replica选举，或者管理员下发分区Leader Replica选举指令。这会在Zookeeper的 /admin/preferred_replica_election指定具体的Topic和分区，此时Leader状态的KafkaController监测到这个路径的数据变化就会触发相应的回调函数，促使对应的Topic分区发生Leader Replica的选举。
OfflinePartitionLeaderSelector：分区状态从OfflinePartition或者NewPartition切换为OnlinePartition时使用的策略

筛选出在线的ISR和AR
优先从在线的ISR中选择，如果列表不为空则选择列表中的第一个，选举结束
在线ISR为空，根据 unclean.leader.election.enable 决定是否从在线的AR中选举Leader，如果允许，则选择AR列表中的第一个，结束选举，如果AR列表为空选举失败。

ControllerShutdownLeaderSelector：Leader状态的KafkaController处理其他Broker Server下线导致分区的Leader Replica发生切换时使用的策略。

筛选出在线的ISR
剔除离线的ISR形成新的ISR列表
如果新的ISR列表不为空，则选举第一个Replica作为新的Leader，否则选举失败

Topic分区的副本状态转换机制

Topic分区的副本状态维护是由ReplicaStateMachine模块负责的，Topic分区的副本状态伴随着Topic分区状态的变化而变化
分区副本状态只要有7种：

NewReplica：分区刚被分配但是没有开始工作的状态
OnlineReplica：分区副本开始工作时的状态，此时该副本时该分区的Leader或者Follower
OfflineReplica：分区副本所在的Broker Server宕机时所导致的副本状态
ReplicaDeletionStarted：分区副本下线之后准备开始删除的状态
ReplicaDeletionSuccessful：相关Broker Server正确响应分区副本被删除请求之后的状态
ReplicaDeletionIneligible：相关Broker Server错误响应分区副本被删除请求之后的状态
NonExistentReplica：代表分区副本被彻底删除之后的状态

目标状态也是由合理的前置状态转换而来的。

KafkaController内部的监听器

KafkaController内部通过监听函数来维护集群的元数据。

TopicChangeListener：注册在 /broker/topics 路径，监听Topic的创建
AddPartitionListener：在Topic创建过程中会在 /broker/topics/[topic]目录下注册AddPartitionListener用于监听Topic分区的变化
PartitionReassignedListener：KafkaController转换为Leader的过程中在路径 /admin/reassign_partitions注册了PartitionReassignedListener用于监听Topic分区的重分配。在正式启动重分配之前会判断是否需要进行重分配，重分配之后的AR列表和当前的AR列表不相同并且重分配之后的AR列表所在的Broker Server都在线，满足上面两个条件才会触发分区的重分配。
ReassignedPartitionsIsrChangeListener：当Leader Replica所在的Broker Server接收到来自Follower Replica的FetchRequest请求时，KafkaApis的handleFetchRequest会统计每个Replica的状态，一旦发现改Replica同步上Leader Replica之后，此时会调用Partition的updateLeaderHWAndMaybeExpandIsr及时更新 /brokers/topics/[topic]/partitions/partitionId/state/ 目录上的Partition状态，包括Leader、ISR等信息，监听到分区状态发生变化会触发ReassignedPartitionsIsrChangeListener
PreferredReplicaElectionListener：每个Partition可以有多个Replica，即AR列表。在这个列表中的第一个Replica称为“Preferred Replica”，当创建Topic时，Kafka要确保所有的Topic的“Preferred Replica”均匀地分布在Kafka集群中。Topic的Partition需要重新均衡Leader Replica至Preferred Replica，此时会触发PreferredReplicaElectionListener
BrokerChangeListener：Broker Server的上下线影响着其中所有Replica的状态，因此ReplicaStateMachine在路径为/broker/topic的路径上注册了BrokerChangeListener，用于监听Broker Server的上下线。
Broker Server上线时步骤为：

Leader状态的KafkaController同步Kafka集群所有的Topic信息给新上线的Broker Server。
将原本位于该Broker Server上的所有Replica状态切换至OnlineReplica
如果Partition的Replica有且只有一个，并且正好位于Broker Server上，则切换Partition状态至OnlinePartition。
如果分区重分配之前由于该Broker Server下线导致推出的话，则尝试重新进行分区重分配。
如果之前由于Broker Server下线导致对应的Replica无法删除的话，则恢复删除流程。
Broker Server下线步骤：
更新ControllerContext内部正在下线的Broker Server列表
将Leader Replica位于该Broker Server上的分区状态切换为OnlinePartition，紧接着触发分区状态切换为OnlinePartition，利用OfflinePartitionLeaderSelector副本选举策略进行Leader Replica的选举。
将位于该Broker Server上的Replica状态切换为OfflineReplica
如果对应Replica的Topic处于删除队列中的话，则标记暂时无法删除。

DeleteTopicsListener：监听Topic的删除

Kafka集群的负载均衡流程

Partition的AR列表的第一个Replica称为“Preferred Replica”，并均匀分布在整个Kafka集群中。由于每个Partition只有Leader Replica对外提供读写服务，并且Partition创建的时候默认的Leader Replica位于Preferred Replica之上，此时Kafka集群的负载是均衡的，如果Kafka集群长时间运行，Broker Server中途由于异常而发生重启，此时Partition的Leader Replica会发生迁移，这样会导致其Partition的Leader Replica在集群中不再均衡了。

Kafka集群的Topic删除流程

Topic是由Partition组成的，而Partition是由Replica组成的，因此只有Partition的Assigned Replica全部被删除了该Partition才可以被删除；只有Topic的所有Partition都被删除了，该Topic才可以最终真正的被删除。

KafkaController的通信模块

ControllerChannelManager提供了Leader状态的KafkaController和集群其他Broker Server通信的功能，内部针对每一个在线的Broker Server会维护一个通信链路，并分别通过各自的RequestSendThread线程将请求发送给对应的Broker Server。

Topic管理工具

kafka-topics.sh提供了Topic的创建、修改、列举、描述、删除功能，在内部时通过TopicCommand类来实现的。
kafka-reassign-partitions.sh提供来重新分配分区副本的能力。该工具可以促进Kafka集群的负载均衡。因为Follower Replica需要从Leader Replica Fetch数据以保持与与Leader Replica同步，仅保持Leader Replica分布的平衡对整个集群的负载均衡时不够的。另外当Kafka集群扩容后，该工具可以将已有Topic的Partition迁移到新加入的Broker上。
分区重分片是一个异步的流程，因此该脚本还提供了查看当前分区重分配进度的指令。
kafka-preferred-replica-election.sh用于在整个集群中恢复Leader Replica为Preferred Replica。

生产者

生产者是指消息的生成者。生产者可以通过特定的分区函数决定消息路由到Topic的某个分区。消息的生产者发送消息有两种模式，分别为同步模式和异步模式。
kafka.javaapi.producer.Producer#send方法发送
指定 metadata.broker.list 属性，配置Broker地址
指定 partitioner.class 属性，配置分区函数，分区函数决定路由。分区函数必须实现 kafka.producer.Partitioner的 partition接口，参数为消息key值，分区总数，返回值为分区的索引。

Producer内部包括以下几个主要模块：

ProducerSendThread：当producer.type配置为async，则其主要用于缓存客户端的KeyedMessage，然后累积到batch.num.messages配置数量或者间隔 queue.enqueue.timeout.ms配置的时间还没有获取到新的客户端的KeyedMessage，则调用DefaultEventHandler将KeyedMessage发送出去
ProducerPool：缓存客户端和各个Broker Server的通信，DefaultEventHandler从ProducerPool中获取和某个Broker Server的通信对象SyncProducer，然后通过SyncProducer将KeyedMessage发送给指定的Broker Server。
DefaultEventHandler：将KeyedMessage集合按照分区规则计算不同Broker Server所应该接收的部分KeyedMessage，然后通过SyncProducer将KeyedMessage发送出去。在DefaultEventHandler模块内部提供来SyncProducer发送失败的重试机制和平滑扩容Broker Server的机制。

发送模式

生产者由两种发送模式：同步和异步
当producer.type配置为sync时，同步发送消息。
当producer.type配置为async时，异步发送消息。

消费者

Kafka提供了两种不同的方式来获取消息：简单消费者和高级消费者。简单消费者获取消息时，用户需要知道待消费的消息位于哪个Topic的哪个分区，并且该目的分区的Leader Replica位于哪个Broker Server上；高级消费者获取消息时，只需要指定待消费的消息属于哪个Topic即可。

简单消费者

简单消费者提供的客户端API称为低级API，本质上客户端获取消息最终时利用FetchRequest请求从目的端Broker Server拉取消息。
FetchRequest请求中可以指定Topic的名称，Topic的分区，起始偏移量、最大字节数。
客户端无论生产消息还是消费消息，最终都是通过与目的地端Broker Server建立通信链路，并且以阻塞模式允许，然后通过该条链路将不同的请求发送出去。

高级消费者

高级消费者以Consumer Group（消费组）的形式来管理消息的消费，以Stream（流）的形式来提供具体消息的读取。Stream是指来自若干个Broker Server上的若干个Partition的消息。客户端需要正确设置Stream的个数，并且应该针对每个Stream开启一个线程进行消息的消费。一个Stream代表了多个Partition消息的聚合，但是每一个Partition只能映射到一个Stream。
消息的最终获取是通过遍历KafkaStream的迭代器ConsumerIterator来逐条获取的，其数据来源于分配给该KafkaStream的阻塞消息队列BlockingQueue，而BlockingQueue的数据来源针对每个Broker Server的FetchThread线程。FetchThread线程会将Broker Server上的部分Partition数据发送给对应的阻塞消息队列BlockingQueue，而KafkaStream正是从该阻塞消息队列BlockingQueue中不断的消费消息。
ConsumerThread本质上是客户端的消费线程，消费若干个Partition上的数据，并且与BlockingQueueu相互映射，只要确定了ConsumerThread和Partition之间的关系，也就确定了BlockingQueue和Partition之间的关系。Kafka提供了两种ConsumerThread和Partition的分配算法Range（范围分区分配）和RoundRobin（循环分区分配）
高级消费者中，每个具体消费者实例启动之后会在/consumers/[group]/ids/的Zookeeper目录下注册自己的id；Kafka集群内部Topic会在/brokers/topics/[topic]/的Zookeeper目录下注册自己的Partition，因此消费者实例一旦发现以上2个路径的数据发生变化时，则会触发高级消费者的负载均衡流程，除此之外，消费者实例一旦和Zookeeper的链接重新建立时也会触发高级消费者的负载均衡流程。
高级消费者内部针对Zookeeper的连接建立、Topic的Partition变化、Consumer的新增会建立3个不同的Listener，分别是ZKSessionExpireListener、ZKTopicPartitionChangeListener和ZKRebalancerListener。
高级消费者消费消息时提供了两种持久化偏移量的机制，由参数auto.commit.enable，默认为true自动提交。否则需要手动调用ZookeeperConsumerConnector的commitOffsets。Kafka根据参数offsets.storage，默认为zookeeper（保存路径为/consumers/[group]/offset/[topic]/[partition]），可以设置为kafka（保存再Topic为“__consumer_offsets”的日志中）。高级消费者内部会自动间隔一定时间（由参数 auto.commit.interval.ms决定，默认60*1000ms）

面试相关

kafka高吞吐量的原因：

一、顺序读写磁盘，充分利用了操作系统的预读机制。
二、linux中使用sendfile命令，减少一次数据拷贝，如下。
①把数据从硬盘读取到内核中的页缓存。
②把数据从内核中读取到用户空间。(sendfile命令将跳过此步骤)
③把用户空间中的数据写到socket缓冲区中。
④操作系统将数据从socket缓冲区中复制到网卡缓冲区，以便将数据经网络发出
三、生产者客户端缓存消息批量发送，消费者批量从broker获取消息，减少网络io次数，充分利用磁盘顺序读写的性能。
四、通常情况下kafka的瓶颈不是cpu或者磁盘，而是网络带宽，所以生产者可以对数据进行压缩。

kafka在高并发的情况下,如何避免消息丢失和消息重复?

消息丢失解决方案:
首先对kafka进行限速，其次启用重试机制，重试间隔时间设置长一些，最后Kafka设置acks=all，即需要相应的所有处于ISR的分区都确认收到该消息后，才算发送成功
消息重复解决方案:
消息可以使用唯一id标识
生产者（ack=all 代表至少成功发送一次)
消费者（offset手动提交，业务逻辑成功处理后，提交offset）
落表（主键或者唯一索引的方式，避免重复数据）
业务逻辑处理（选择唯一主键存储到Redis或者mongdb中，先查询是否存在，若存在则不处理；若不存在，先插入Redis或Mongdb,再进行业务逻辑处理）

kafka怎么保证数据消费一次且仅消费一次

幂等producer：保证发送单个分区的消息只会发送一次，不会出现重复消息
事务(transaction)：保证原子性地写入到多个分区，即写入到多个分区的消息要么全部成功，要么全部回滚流处理EOS：流处理本质上可看成是“读取-处理-写入”的管道。此EOS保证整个过程的操作是原子性。注意，这只适用于Kafka Streams

kafka保证数据一致性和可靠性

数据一致性保证

一致性定义：若某条消息对client可见，那么即使Leader挂了，在新Leader上数据依然可以被读到
HW-HighWaterMark: client可以从Leader读到的最大msg offset，即对外可见的最大offset， HW=max(replica.offset)
对于Leader新收到的msg，client不能立刻消费，Leader会等待该消息被所有ISR中的replica同步后，更新HW，此时该消息才能被client消费，这样就保证了如果Leader fail，该消息仍然可以从新选举的Leader中获取。
对于来自内部Broker的读取请求，没有HW的限制。同时，Follower也会维护一份自己的HW，Folloer.HW = min(Leader.HW, Follower.offset)

数据可靠性保证

当Producer向Leader发送数据时，可以通过acks参数设置数据可靠性的级别
0: 不论写入是否成功，server不需要给Producer发送Response，如果发生异常，server会终止连接，触发Producer更新meta数据；
1: Leader写入成功后即发送Response，此种情况如果Leader fail，会丢失数据
-1: 等待所有ISR接收到消息后再给Producer发送Response，这是最强保证

kafka到spark streaming怎么保证数据完整性，怎么保证数据不重复消费？

保证数据不丢失（at-least）

spark RDD内部机制可以保证数据at-least语义。
Receiver方式
开启WAL（预写日志），将从kafka中接受到的数据写入到日志文件中，所有数据从失败中可恢复。
Direct方式
依靠checkpoint机制来保证。

保证数据不重复（exactly-once）

要保证数据不重复，即Exactly once语义。

幂等操作：重复执行不会产生问题，不需要做额外的工作即可保证数据不重复。
业务代码添加事务操作
就是说针对每个partition的数据，产生一个uniqueId，只有这个partition的所有数据被完全消费，则算成功，否则算失效，要回滚。下次重复执行这个uniqueId时，如果已经被执行成功，则skip掉。