Java重要面试名词整理（十）：Kafka

文章目录

- Kafka简介
- - 相关概念
  - Kraft集群
- Kafka收发消息
- - 梳理客户端工作机制
  - - 消费者分组消费机制
    - 生产者拦截器机制
    - 消息序列化机制
    - 消息分区路由机制
    - 生产者消息缓存机制
    - 发送应答机制
    - 生产者消息幂等性
    - 生产者消息事务
- Kafka集群架构设计-Kafka的Zookeeper元数据梳理
- - Leader Partition选举机制
  - Leader Partition自动平衡机制
  - Partition故障恢复机制
  - HW一致性保障-Epoch更新机制
- Kafka的文件高效读写机制
- - - 零拷贝
- Kafka生产调优

Kafka简介

Kafka是一个分布式的发布-订阅消息系统，可以快速地处理高吞吐量的数据流，并将数据实时地分发到多个消费者中。Kafka消息系统由多个broker（服务器）组成，这些broker可以在多个数据中心之间分布式部署，以提供高可用性和容错性。

Kafka的基本架构由生产者、消费者和主题（topic）组成。生产者可以将数据发布到指定的主题，而消费者可以订阅这些主题并消费其中的数据。同时，Kafka还支持数据流的处理和转换，可以在管道中通过Kafka Streams API进行流式计算，例如过滤、转换、聚合等。

Kafka在企业级应用中被广泛应用，包括实时流处理、日志聚合、监控和数据分析等方面。同时，Kafka还可以与其他大数据工具集成，如Hadoop、Spark和Storm等，构建一个完整的数据处理生态系统。

Kraft集群

Kraft是Kafka从2.8.0版本开始支持的一种新的集群架构方式。其目的主要是为了摆脱Kafka对Zookeeper的依赖。因为以往基于Zookeeper搭建的集群，增加了Kafka演进与运维的难度，逐渐开始成为Kakfa拥抱云原生的一种障碍。使用Kraft集群后，Kafka集群就不再需要依赖Zookeeper，将之前基于Zookeeper管理的集群数据，转为由Kafka集群自己管理。

Raft协议是目前进行去中心化集群管理的一种常见算法，类似于之前的Paxos协议，是一种基于多数同意，从而产生集群共识的分布式算法。Kraft则是Kafka基于Raft协议进行的定制算法。

新的Kraft集群相比传统基于Zookeeper的集群，有一些很明显的好处：

Kafka可以不依赖于外部框架独立运行。这样减少Zookeeper性能抖动对Kafka集群性能的影响，同时Kafka产品的版本迭代也更自由。
Controller不再由Zookeeper动态选举产生，而是由配置文件进行固定。这样比较适合配合一些高可用工具来保持集群的稳定性。
Zookeeper的产品特性决定了他不适合存储大量的数据，这对Kafka的集群规模(确切的说应该是Partition规模)是极大的限制。摆脱Zookeeper后，集群扩展时元数据的读写能力得到增强。

不过，由于分布式算法的复杂性。Kraft集群和同样基于Raft协议定制的RocketMQ的Dledger集群一样，都还在不太稳定，在真实企业开发中，用得相对还是比较少。

Kafka收发消息

Kafka提供了两套客户端API，HighLevel API和LowLevel API。 HighLevel API封装了kafka的运行细节，使用起来比较简单，是企业开发过程中最常用的客户端API。而LowLevel API则需要客户端自己管理Kafka的运行细节，Partition，Offset这些数据都由客户端自行管理。这层API功能更灵活，但是使用起来非常复杂，也更容易出错。只在极少数对性能要求非常极致的场景才会偶尔使用。我们的重点是HighLevel API 。

整体来说，构建Producer分为三个步骤：

设置Producer核心属性 ：Producer可选的属性都可以由ProducerConfig类管理。比如ProducerConfig.BOOTSTRAP_SERVERS_CONFIG属性，显然就是指发送者要将消息发到哪个Kafka集群上。这是每个Producer必选的属性。在ProducerConfig中，对于大部分比较重要的属性，都配置了对应的DOC属性进行描述。
构建消息：Kafka的消息是一个Key-Value结构的消息。其中，key和value都可以是任意对象类型。其中，key主要是用来进行Partition分区的，业务上更关心的是value。
使用Producer发送消息。：通常用到的就是单向发送、同步发送和异步发送者三种发送方式。

整体来说，Consumer同样是分为三个步骤：

设置Consumer核心属性 ：可选的属性都可以由ConsumerConfig类管理。在这个类中，同样对于大部分比较重要的属性，都配置了对应的DOC属性进行描述。同样BOOTSTRAP_SERVERS_CONFIG是必须设置的属性。
拉取消息：Kafka采用Consumer主动拉取消息的Pull模式。consumer主动从Broker上拉取一批感兴趣的消息。
处理消息，提交位点：消费者将消息拉取完成后，就可以交由业务自行处理对应的这一批消息了。只是消费者需要向Broker提交偏移量offset。如果不提交Offset，Broker会认为消费者端消息处理失败了，还会重复进行推送。

Kafka的客户端基本就是固定的按照这三个大的步骤运行。在具体使用过程中，最大的变数基本上就是给生产者和消费者的设定合适的属性。这些属性极大的影响了客户端程序的执行方式。

kafka官方配置： https://kafka.apache.org/documentation/#configuration。

Kafka的设计精髓，是在网络不稳定，服务也随时会崩溃的这些作死的复杂场景下，如何保证消息的高并发、高吞吐，那才是Kafka最为精妙的地方。但是要理解那些复杂的问题，都是需要建立在这个基础模型基础上的。

梳理客户端工作机制

消费者分组消费机制

这是我们在使用kafka时，最为重要的一个机制，因此最先进行梳理。

生产者往Topic下发消息时，会尽量均匀的将消息发送到Topic下的各个Partition当中。而这个消息，会向所有订阅了该Topic的消费者推送。推送时，每个ConsumerGroup中只会推送一份。也就是同一个消费者组中的多个消费者实例，只会共同消费一个消息副本。而不同消费者组之间，会重复消费消息副本。这就是消费者组的作用。

与之相关的还有Offset偏移量。这个偏移量表示每个消费者组在每个Partiton中已经消费处理的进度。在Kafka中，可以看到消费者组的Offset记录情况。

这个Offset偏移量，需要消费者处理完成后主动向Kafka的Broker提交。提交完成后，Broker就会更新消费进度，表示这个消息已经被这个消费者组处理完了。但是如果消费者没有提交Offset，Broker就会认为这个消息还没有被处理过，就会重新往对应的消费者组进行推送，不过这次，一般会尽量推送给同一个消费者组当中的其他消费者实例。

生产者拦截器机制

生产者拦截机制允许客户端在生产者在消息发送到Kafka集群之前，对消息进行拦截，甚至可以修改消息内容。

拦截器机制一般用得比较少，主要用在一些统一添加时间等类似的业务场景。比如，用Kafka传递一些POJO，就可以用拦截器统一添加时间属性。但是我们平常用Kafka传递的都是String类型的消息，POJO类型的消息，Kafka可以传吗？这就要用到下面的消息序列化机制。

消息序列化机制

Producer指定了两个属性KEY_SERIALIZER_CLASS_CONFIG和VALUE_SERIALIZER_CLASS_CONFIG，对于这两个属性，在ProducerConfig中都有配套的说明属性。

通过这两个参数，可以指定消息生产者如何将消息的key和value序列化成二进制数据。在Kafka的消息定义中，key和value的作用是不同的。

key是用来进行分区的可选项。Kafka通过key来判断消息要分发到哪个Partition。

Value是业务上比较关心的消息。Kafka同样需要将Value对象通过Serializer序列化接口，将value转换成byte[]数组，这样才能比较好的在网络上传输Value信息，以及将Value信息落盘到操作系统的文件当中。

生产者要对消息进行序列化，那么消费者拉取消息时，自然需要进行反序列化

序列化机制是在高并发场景中非常重要的一个优化机制。高效的序列化能够极大的提升分布式系统的网络传输以及数据落盘的能力。例如对于一个User对象，即可以使用JSON字符串这种简单粗暴的序列化方式，也可以选择按照各个字段进行组合序列化的方式。但是显然后者的占用空间比较小，序列化速度也会比较快。而Kafka在文件落盘时，也设计了非常高效的数据序列化实现，这也是Kafka高效运行的一大支撑。

在很多其他业务场景中，也需要我们提供更高效的序列化实现。例如使用MapReduce框架时，就需要自行定义数据的序列化方式。使用Netty框架进行网络调用时，为了防止粘包，也需要定制数据的序列化机制。在这些场景下，进行序列化的基础思想，和我们这里介绍的也是一样的。

消息分区路由机制

在Producer中，可以指定一个Partitioner来对消息进行分配。然后，在Consumer中，可以指定一个PARTITION_ASSIGNMENT_STRATEGY分区分配策略，决定如何在多个Consumer实例和多个Partitioner之间建立关联关系。

Kafka默认提供了三种消费者的分区分配策略

range策略：比如一个Topic有10个Partiton(partition 0~9) 一个消费者组下有三个Consumer(consumer1~3)。Range策略就会将分区0~3分给一个Consumer，4~6给一个Consumer，7~9给一个Consumer。
round-robin策略：轮询分配策略，可以理解为在Consumer中一个一个轮流分配分区。比如0，3，6，9分区给一个Consumer，1，4，7分区给一个Consumer，然后2，5，8给一个Consumer
sticky策略：粘性策略。这个策略有两个原则：
- 1、在开始分区时，尽量保持分区的分配均匀。比如按照Range策略分(这一步实际上是随机的)。
- 2、分区的分配尽可能的与上一次分配的保持一致。比如在range分区的情况下，第三个Consumer的服务宕机了，那么按照sticky策略，就会保持consumer1和consumer2原有的分区分配情况。然后将consumer3分配的7~9分区尽量平均的分配到另外两个consumer上。这种粘性策略可以很好的保持Consumer的数据稳定性。

另外可以通过继承AbstractPartitionAssignor抽象类自定义消费者的订阅方式。

官方默认提供的生产者端的默认分区器以及消费者端的RangeAssignor+CooperativeStickyAssignor分配策略，在大部分场景下都是非常高效的算法。深入理解这些算法，对于你深入理解MQ场景，以及借此去横向对比理解其他的MQ产品，都是非常有帮助的。

生产者消息缓存机制

Kafka生产者为了避免高并发请求对服务端造成过大压力，每次发消息时并不是一条一条发往服务端，而是增加了一个高速缓存，将消息集中到缓存后，批量进行发送。这种缓存机制也是高并发处理时非常常用的一种机制。

Kafka的消息缓存机制涉及到KafkaProducer中的两个关键组件： accumulator 和 sender

RecordAccumulator，就是Kafka生产者的消息累加器。KafkaProducer要发送的消息都会在ReocrdAccumulator中缓存起来，然后再分批发送给kafka broker。

sender就是KafkaProducer中用来发送消息的一个单独的线程。每个KafkaProducer对象都对应一个sender线程。他会负责将RecordAccumulator中的消息发送给Kafka。

Sender也并不是一次就把RecordAccumulator中缓存的所有消息都发送出去，而是每次只拿一部分消息。

然后，Sender对读取出来的消息，会以Broker为key，缓存到一个对应的队列当中。这些队列当中的消息就称为InflightRequest。接下来这些Inflight就会一一发往Kafka对应的Broker中，直到收到Broker的响应，才会从队列中移除。这些队列也并不会无限缓存，最多缓存MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION(默认值为5)个请求。

生产者缓存机制的主要目的是将消息打包，减少网络IO频率。所以，在Sender的InflightRequest队列中，消息也不是一条一条发送给Broker的，而是一批消息一起往Broker发送。而这就意味着这一批消息是没有固定的先后顺序的。

发送应答机制

Producer有一个属性ACKS_CONFIG。

如果你理解了Topic的分区模型，这个属性就非常容易理解了。这个属性更大的作用在于保证消息的安全性，尤其在replica-factor备份因子比较大的Topic中，尤为重要。

acks=0，生产者不关心Broker端有没有将消息写入到Partition，只发送消息就不管了。吞吐量是最高的，但是数据安全性是最低的。
acks=all or -1，生产者需要等Broker端的所有Partiton(Leader Partition以及其对应的Follower Partition都写完了才能得到返回结果，这样数据是最安全的，但是每次发消息需要等待更长的时间，吞吐量是最低的。
acks设置成1，则是一种相对中和的策略。Leader Partition在完成自己的消息写入后，就向生产者返回结果。

生产者消息幂等性

上面分析过，当Producer的acks设置成1或-1时，Producer每次发送消息都是需要获取Broker端返回的RecordMetadata的。这个过程中就需要两次跨网络请求。

如果要保证消息安全，那么对于每个消息，这两次网络请求就必须要求是幂等的。但是，网络是不靠谱的，在高并发场景下，往往没办法保证这两个请求是幂等的。Producer发送消息的过程中，如果第一步请求成功了，但是第二步却没有返回。这时，Producer就会认为消息发送失败了。那么Producer必然会发起重试。重试次数由参数ProducerConfig.RETRIES_CONFIG，默认值是Integer.MAX。

Producer会重复发送多条消息到Broker中。Kafka如何保证无论Producer向Broker发送多少次重复的数据，Broker端都只保留一条消息，而不会重复保存多条消息呢？这就是Kafka消息生产者的幂等性问题。

这里首先需要理解分布式数据传递过程中的三个数据语义：at-least-once:至少一次；at-most-once:最多一次；exactly-once:精确一次。

举个例子，你往银行存100块钱，这时银行往往需要将存钱动作转化成一个消息，发到MQ，然后通过MQ通知另外的系统去完成修改你的账户余额以及其他一些其他的业务动作。而这个MQ消息的安全性，往往是需要分层次来设计的。首先，你要保证存钱的消息能够一定发送到MQ。如果一次发送失败了，那就重试几次，只到成功为止。这就是at-least-once至少一次。如果保证不了这个语义，那么你肯定不会接受。然后，你往银行存100块，不管这个消息你发送了多少次，银行最多只能记录一次，也就是100块存款，可以少，但决不能多。这就是at-most-once最多一次。如果保证不了这个语义，那么银行肯定也不能接受。最后，这个业务动作要让双方都满意，就必须保证存钱这个消息正正好好被记录一次，不多也不少。这就是Exactly-once语义。

回到Producer发消息给Broker这个场景，如果要保证at-most-once语义，可以将ack级别设置为0即可，此时，是不存在幂等性问题的。如果要保证at-least-once语义，就需要将ack级别设置为1或者-1，这样就能保证Leader Partition中的消息至少是写成功了一次的，但是不保证只写了一次。如果要支持Exactly-once语义怎么办呢？这就需要使用到idempotence幂等性属性了。

Kafka为了保证消息发送的Exactly-once语义，增加了几个概念：

PID：每个新的Producer在初始化的过程中就会被分配一个唯一的PID。这个PID对用户是不可见的。
Sequence Numer: 对于每个PID，这个Producer针对Partition会维护一个sequenceNumber。这是一个从0开始单调递增的数字。当Producer要往同一个Partition发送消息时，这个Sequence Number就会加1。然后会随着消息一起发往Broker。

Broker端则会针对每个<PID,Partition>维护一个序列号（SN），只有当对应的SequenceNumber = SN+1时，Broker才会接收消息，同时将SN更新为SN+1。否则，SequenceNumber过小就认为消息已经写入了，不需要再重复写入。而如果SequenceNumber过大，就会认为中间可能有数据丢失了。对生产者就会抛出一个OutOfOrderSequenceException。

这样，Kafka在打开idempotence幂等性控制后，在Broker端就会保证每条消息在一次发送过程中，Broker端最多只会刚刚好持久化一条。这样就能保证at-most-once语义。再加上之前分析的将生产者的acks参数设置成1或-1，保证at-least-once语义，这样就整体上保证了Exactaly-once语义

生产者消息事务

通过生产者消息幂等性问题，能够解决单生产者消息写入单分区的的幂等性问题。但是，如果是要写入多个分区呢？

这时候，通过上面的生产者消息幂等性机制就无法保证所有消息的幂等了。这时候就需要有一个事务机制，保证这一批消息最好同时成功的保持幂等性。或者这一批消息同时失败，这样生产者就可以开始进行整体重试，消息不至于重复。

而针对这个问题， Kafka就引入了消息事务机制。这涉及到Producer中的几个API：

// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 提交事务
void commitTransaction() throws ProducerFencedException;
// 4 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;

Kafka的事务消息还会做两件事情：

1、一个TransactionId只会对应一个PID

如果当前一个Producer的事务没有提交，而另一个新的Producer保持相同的TransactionId，这时旧的生产者会立即失效，无法继续发送消息。

2、跨会话事务对齐

如果某个Producer实例异常宕机了，事务没有被正常提交。那么新的TransactionId相同的Producer实例会对旧的事务进行补齐。保证旧事务要么提交，要么终止。这样新的Producer实例就可以以一个正常的状态开始工作。

对于事务ID这个参数，可以任意起名，但是建议包含一定的业务唯一性。

生产者的事务消息机制保证了Producer发送消息的安全性，但是，他并不保证已经提交的消息就一定能被所有消费者消费。

Kafka集群架构设计-Kafka的Zookeeper元数据梳理

Kafka依赖很多的存储数据，但是，总体上是有划分的。Kafka会将每个服务的不同之处，也就是状态信息，保存到Zookeeper中。通过Zookeeper中的数据，指导每个Kafka进行与其他Kafka节点不同的业务逻辑。而将状态信息抽离后，剩下的数据，就可以直接存在Kafka本地，所有Kafka服务都以相同的逻辑运行。这种状态信息分离的设计，让Kafka有非常好的集群扩展性。

Leader Partition选举机制

AR: Assigned Repllicas。表示Kafka分区中的所有副本(存活的和不存活的)
ISR: 表示在所有AR中，服务正常，保持与Leader同步的Follower集合。如果Follower长时间没有向Leader发送通信请求(超时时间由replica.lag.time.max.ms参数设定，默认30S)，那么这个Follower就会被踢出ISR中。(在老版本的Kafka中，还会考虑Partition与Leader Partition之间同步的消息差值，大于参数replica.lag.max.messages条就会被移除ISR。现在版本已经移除了这个参数。)
OSR：表示从ISR中踢出的节点。记录的是那些服务有问题，延迟过多的副本。

Leader Partition自动平衡机制

Kafka设计了Leader Partition自动平衡机制，当发现Leader分配不均衡时，自动进行Leader Partition调整。

Kafka在进行Leader Partition自平衡时的逻辑是这样的：他会认为AR当中的第一个节点就应该是Leader节点。这种选举结果成为preferred election 理想选举结果。Controller会定期检测集群的Partition平衡情况，在开始检测时，Controller会依次检查所有的Broker。当发现这个Broker上的不平衡的Partition比例高于leader.imbalance.per.broker.percentage阈值时，就会触发一次Leader Partiton的自平衡。

Partition故障恢复机制

Kafka为了保证消息能够在多个Parititon中保持数据同步，内部记录了两个关键的数据：

LEO(Log End Offset): 每个Partition的最后一个Offset

这个参数比较好理解，每个Partition都会记录自己保存的消息偏移量。leader partition收到并记录了生产者发送的一条消息，就将LEO加1。而接下来，follower partition需要从leader partition同步消息，每同步到一个消息，自己的LEO就加1。通过LEO值，就知道各个follower partition与leader partition之间的消息差距。

HW(High Watermark): 一组Partiton中最小的LEO。

follower partition每次往leader partition同步消息时，都会同步自己的LEO给leader partition。这样leader partition就可以计算出这个HW值，并最终会同步给各个follower partition。leader partition认为这个HW值以前的消息，都是在所有follower partition之间完成了同步的，是安全的。这些安全的消息就可以被消费者拉取过去了。而HW值之后的消息，就是不安全的，是可能丢失的。这些消息如果被消费者拉取过去消费了，就有可能造成数据不一致。

HW一致性保障-Epoch更新机制

Kafka设计了Epoch机制，来保证HW的一致性。

Epoch是一个单调递增的版本号，每当Leader Partition发生变更时，该版本号就会更新。所以，当有多个Epoch时，只有最新的Epoch才是有效的，而其他Epoch对应的Leader Partition就是过期的，无用的Leader。
每个Leader Partition在上任之初，都会新增一个新的Epoch记录。这个记录包含更新后端的epoch版本号，以及当前Leader Partition写入的第一个消息的偏移量。例如(1,100)。表示epoch版本号是1，当前Leader Partition写入的第一条消息是100. Broker会将这个epoch数据保存到内存中，并且会持久化到本地一个leader-epoch-checkpoint文件当中。
这个leader-epoch-checkpoint会在所有Follower Partition中同步。当Leader Partition有变更时，新的Leader Partition就会读取这个Epoch记录，更新后添加自己的Epoch记录。
接下来其他Follower Partition要更新数据时，就可以不再依靠自己记录的HW值判断拉取消息的起点。而可以根据这个最新的epoch条目来判断。

Kafka的文件高效读写机制

零拷贝

零拷贝是Linux操作系统提供的一种IO优化机制，而Kafka大量的运用了零拷贝机制来加速文件读写。

内核态的内容复制是在内核层面进行的，而零拷贝的技术，重点是要配合内核态的复制机制，减少用户态与内核态之间的内容拷贝。

具体实现时有两种方式：

1、mmap文件映射机制

这种方式是在用户态不再缓存整个IO的内容，改为只持有文件的一些映射信息。通过这些映射，"遥控"内核态的文件读写。这样就减少了内核态与用户态之间的拷贝数据大小，提升了IO效率。

mmap文件映射机制是操作系统提供的一种文件操作机制，可以使用man 2 mmap查看。实际上在Java程序执行过程当中就会被大量使用。

2、sendfile文件传输机制

这种机制可以理解为用户态，也就是应用程序不再关注数据的内容，只是向内核态发一个sendfile指令，要他去复制文件就行了。这样数据就完全不用复制到用户态，从而实现了零拷贝。

相比mmap，连索引都不读了，直接通知操作系统去拷贝就是了。

JDK中8中java.nio.channels.FileChannel类提供了transferTo和transferFrom方法，底层就是使用了操作系统的sendfile机制。

这些底层的优化机制都是操作系统提供的优化机制，其实针对任何上层应用语言来说，都是一个黑盒，只能去调用，但是控制不了具体的实现过程。而上层的各种各样的语言，也只能根据操作系统提供的支持进行自己的实现。虽然不同语言的实现方式会有点不同，但是本质都是一样的。

Kafka生产调优

生产环境通常会对Kafka搭建监控平台。而Kafka-eagle就是一个可以监控Kafka集群整体运行情况的框架，在生产环境经常会用到。官网地址：https://www.kafka-eagle.org/ 以前叫做Kafka-eagle，现在用了个简写，EFAK（Eagle For Apache Kafka）