Kafka关键原理加强及一些面试题-Toy模板网

这篇具有很好参考价值的文章主要介绍了Kafka关键原理加强及一些面试题。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

发文章是对自己所学的内容做个记录, 同时看看能否帮助更多需要帮助的人 .

6.kafka关键原理加强

6.1日志分段切分条件

日志分段文件切分包含以下4个条件，满足其一即可：

当前日志分段文件的大小超过了broker端参数 log.segment.bytes 配置的值。

log.segment.bytes参数的默认值为 1073741824，即1GB

当前日志分段中消息的最小时间戳与当前系统的时间戳的差值大于log.roll.ms或log.roll.hours参数配置的值。如果同时配置了log.roll.ms和log.roll.hours 参数，那么 log.roll.ms 的优先级高，默认情况下，只配置了log.roll.hours参数，其值为168,即7天。

偏移量索引文件或时间戳索引文件的大小达到 broker 端参数 log.index.size.max.bytes 配置的值。

log.index.size .max.bytes的默认值为 10485760，即10MB

追加的消息的偏移量与当前日志分段的起始偏移量之间的差值大于Integer.MAX_VALUE, 即要追加的消息的偏移量不能转变为相对偏移量（offset - baseOffset > Integer.MAX_VALUE）。

6.2什么是Controller

Controller作为Kafka集群中的核心组件，它的主要作用是在 Apache ZooKeeper 的帮助下管理和协调整个 Kafka 集群。

Controller与Zookeeper进行交互，获取与更新集群中的元数据信息。其他broker并不直接与zookeeper进行通信，而是与 Controller 进行通信并同步Controller中的元数据信息。

Kafka集群中每个节点都可以充当Controller节点，但集群中同时只能有一个Controller节点。

Controller简单来说，就是kafka集群的状态管理者

controller竞选机制：简单说，先来先上！

Broker 在启动时，会尝试去 ZooKeeper 中创建 /controller 节点。Kafka 当前选举控制器的规则是：第一个成功创建 /controller 节点的 Broker 会被指定为控制器。

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器（Kafka Controller），它负责维护整个集群中所有分区和副本的状态及分区leader的选举。

当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责分区的重新分配。

Kafka中的控制器选举的工作依赖于Zookeeper，成功竞选为控制器的broker会在Zookeeper中创建/controller这个临时（EPHEMERAL）节点，此临时节点的内容参考如下：

{"version":1,"brokerid":0,"timestamp":"1529210278988"}

其中version在目前版本中固定为1，brokerid表示成为控制器的broker的id编号，timestamp表示竞选成为控制器时的时间戳。

在任意时刻，集群中有且仅有一个控制器。每个broker启动的时候会去尝试去读取zookeeper上的/controller节点的brokerid的值，如果读取到brokerid的值不为-1，则表示已经有其它broker节点成功竞选为控制器，所以当前broker就会放弃竞选；如果Zookeeper中不存在/controller这个节点，或者这个节点中的数据异常，那么就会尝试去创建/controller这个节点，当前broker去创建节点的时候，也有可能其他broker同时去尝试创建这个节点，只有创建成功的那个broker才会成为控制器，而创建失败的broker则表示竞选失败。每个broker都会在内存中保存当前控制器的brokerid值，这个值可以标识为activeControllerId。

6.2.1controller的职责

监听partition相关变化

TypeScript
对Zookeeper中的/admin/reassign_partitions节点注册PartitionReassignmentListener，用来处理分区重分配的动作。
对Zookeeper中的/isr_change_notification节点注册IsrChangeNotificetionListener，用来处理ISR集合变更的动作。
对Zookeeper中的/admin/preferred-replica-election节点添加PreferredReplicaElectionListener，用来处理优先副本选举。

监听topic增减变化

Plain Text
对Zookeeper中的/brokers/topics节点添加TopicChangeListener，用来处理topic增减的变化；
对Zookeeper中的/admin/delete_topics节点添加TopicDeletionListener，用来处理删除topic的动作。

监听broker相关的变化

Plain Text
对Zookeeper中的/brokers/ids/节点添加BrokerChangeListener，用来处理broker增减的变化。

更新集群的元数据信息

Plain Text
从Zookeeper中读取获取当前所有与topic、partition以及broker有关的信息并进行相应的管理。对各topic所对应的Zookeeper中的/brokers/topics/[topic]节点添加PartitionModificationsListener，用来监听topic中的分区分配变化。并将最新信息同步给其他所有broker。

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

如果参数auto.leader.rebalance.enable设置为true，则还会开启一个名为“auto-leader-rebalance-task”的定时任务来负责维护分区的leader副本的均衡。

6.2.2分区的负载分布

客户端请求创建一个topic时，每一个分区副本在broker上的分配，是由集群controller来决定；

结论：里面会创建出来两个随机数

第一个随机数确定0号分区leader的位置，往后1号分区2号分区的leader依次往后顺延1

第二个随机数确定每个分区的第一个副本的位置在leader所在机器上往后顺延（随机数+1）台机器，

该台机器就是第一个副本的位置，剩余副本依次往后顺延1

举例：

broker_id = 0~19 一共20台机器

分区数20,副本数10

第一个随机数：19

第二个随机数：0

(0,ArrayBuffer(19, 0, 1, 2, 3, 4, 5, 6, 7, 8))

(1,ArrayBuffer(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))

(2,ArrayBuffer(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))

(3,ArrayBuffer(2, 3, 4, 5, 6, 7, 8, 9, 10, 11))

(4,ArrayBuffer(3, 4, 5, 6, 7, 8, 9, 10, 11, 12))

(5,ArrayBuffer(4, 5, 6, 7, 8, 9, 10, 11, 12, 13))

(6,ArrayBuffer(5, 6, 7, 8, 9, 10, 11, 12, 13, 14))

(7,ArrayBuffer(6, 7, 8, 9, 10, 11, 12, 13, 14, 15))

(8,ArrayBuffer(7, 8, 9, 10, 11, 12, 13, 14, 15, 16))

(9,ArrayBuffer(8, 9, 10, 11, 12, 13, 14, 15, 16, 17))

(10,ArrayBuffer(9, 10, 11, 12, 13, 14, 15, 16, 17, 18))

(11,ArrayBuffer(10, 11, 12, 13, 14, 15, 16, 17, 18, 19))

(12,ArrayBuffer(11, 12, 13, 14, 15, 16, 17, 18, 19, 0))

(13,ArrayBuffer(12, 13, 14, 15, 16, 17, 18, 19, 0, 1))

(14,ArrayBuffer(13, 14, 15, 16, 17, 18, 19, 0, 1, 2))

(15,ArrayBuffer(14, 15, 16, 17, 18, 19, 0, 1, 2, 3))

(16,ArrayBuffer(15, 16, 17, 18, 19, 0, 1, 2, 3, 4))

(17,ArrayBuffer(16, 17, 18, 19, 0, 1, 2, 3, 4, 5))

(18,ArrayBuffer(17, 18, 19, 0, 1, 2, 3, 4, 5, 6))

(19,ArrayBuffer(18, 19, 0, 1, 2, 3, 4, 5, 6, 7))

其分布策略源码如下：

Java
private def assignReplicasToBrokersRackUnaware(
nPartitions: Int, //分区的个数   10
replicationFactor: Int, //副本的个数 5
brokerList: Seq[Int],//broker的集合    8   0~7
fixedStartIndex: Int//默认值是-1 固定开始的索引位置
startPartitionId: Int): Map[Int, Seq[Int]] //默认值是-1 分区开始的位置
= {
val ret = mutable.Map[Int, Seq[Int]]()
val brokerArray = brokerList.toArray
val startIndex = if (fixedStartIndex >= 0) {
      fixedStartIndex
}else {
          rand.nextInt(brokerArray.length)
}
var currentPartitionId = math.max(0, startPartitionId)
var nextReplicaShift = if (fixedStartIndex >= 0) {
          fixedStartIndex
}else {
          rand.nextInt(brokerArray.length)
}
for (_ <- 0 until nPartitions) {
    if (currentPartitionId > 0 && (currentPartitionId % brokerArray.length == 0)){
      nextReplicaShift += 1
        }

    val firstReplicaIndex = (currentPartitionId + startIndex) % brokerArray.length
    val replicaBuffer = mutable.ArrayBuffer(brokerArray(firstReplicaIndex))
    for (j <- 0 until replicationFactor - 1) {
      replicaBuffer += brokerArray(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerArray.length))
    }
    ret.put(currentPartitionId, replicaBuffer)
    currentPartitionId += 1
}
ret
}

private def replicaIndex(firstReplicaIndex: Int, secondReplicaShift: Int, replicaIndex: Int, nBrokers: Int): Int = {
val shift = 1 + (secondReplicaShift + replicaIndex) % (nBrokers - 1)
(firstReplicaIndex + shift) % nBrokers
}

副本因子不能大于 Broker 的个数；

报错：Replication factor: 4 larger than available brokers: 3.

partition_0的第1个副本（leader副本）放置位置是随机从 brokerList 选择的；

其他分区的第1个副本（leader）放置位置相对于paritition_0分区依次往后移（也就是如果我们有5个 Broker，5个分区，假设partition0分区放在broker4上，那么partition1将会放在broker5上；patition2将会放在broker1上；partition3在broker2，依次类）；

各分区剩余的副本相对于分区前一个副本偏移随机数nextReplicaShift+1，然后后面的副本依次加1

6.2.3分区Leader的选举机制

分区 leader 副本的选举由控制器controller负责具体实施。

当创建分区（创建主题或增加分区都有创建分区的动作）或Leader下线（此时分区需要选举一个新的leader上线来对外提供服务）的时候都需要执行 leader 的选举动作。

选举策略：按照 ISR集合中副本的顺序查找第一个存活的副本，并且这个副本在 ISR 集合中

一个分区的AR集合在partition分配的时候就被指定，并且只要不发生重分配的情况，集合内部副本的顺序是保持不变的，而分区的 ISR 集合中副本的顺序可能会改变；

6.3生产者原理解析

生产者工作流程图：

一个生产者客户端由两个线程协调运行，这两个线程分别为主线程和 Sender 线程。

在主线程中由kafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器（RecordAccumulator, 也称为消息收集器）中。

Sender 线程负责从RecordAccumulator 获取消息并将其发送到 Kafka 中；

RecordAccumulator主要用来缓存消息以便Sender 线程可以批量发送，进而减少网络传输的资源消耗以提升性能。RecordAccumulator缓存的大小可以通过生产者客户端参数buffer.memory 配置，默认值为 33554432B ，即32M。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候 KafkaProducer.send（）方法调用要么被阻塞，要么抛出异常，这个取决于参数 max.block.ms 的配置，此参数的默认值为 60000,即60秒。

主线程中发送过来的消息都会被迫加到 RecordAccumulator 的某个双端队列（ Deque ）中，

RecordAccumulator内部为每个分区都维护了一个双端队列，即Deque<ProducerBatch>。

消息写入缓存时，追加到双端队列的尾部;

Sender读取消息时，从双端队列的头部读取。注意：ProducerBatch 是指一个消息批次；

与此同时，会将较小的 ProducerBatch 凑成一个较大 ProducerBatch ，也可以减少网络请求的次数以提升整体的吞吐量。

ProducerBatch 大小和 batch.size 参数也有着密切的关系。当一条消息（ProducerRecord ) 流入 RecordAccumulator 时，会先寻找与消息分区所对应的双端队列（如果没有则新建），再从这个双端队列的尾部获取一个ProducerBatch （如果没有则新建），查看 ProducerBatch中是否还可以写入这个ProducerRecord，如果可以写入就直接写入，如果不可以则需要创建一个新的Producer Batch。在新建 ProducerBatch时评估这条消息的大小是否超过 batch.size 参数大小，如果不超过，那么就以 batch.size 参数的大小来创建 ProducerBatch。

如果生产者客户端需要向很多分区发送消息，则可以将buffer.memory参数适当调大以增加整体的吞吐量。

Sender从 RecordAccumulator 获取缓存的消息之后，会进一步将<分区,Deque<Producer Batch>>的形式转变成<Node,List< ProducerBatch>的形式，其中Node表示Kafka集群broker节点。对于网络连接来说，生产者客户端是与具体broker节点建立的连接，也就是向具体的broker节点发送消息，而并不关心消息属于哪一个分区；而对于KafkaProducer的应用逻辑而言，我们只关注向哪个分区中发送哪些消息，所以在这里需要做一个应用逻辑层面到网络I/O层面的转换。

在转换成<Node, List<ProducerBatch>>的形式之后， Sender会进一步封装成<Node,Request> 的形式，这样就可以将 Request 请求发往各个Node了，这里的Request是Kafka各种协议请求；

请求在从sender线程发往Kafka之前还会保存到InFlightRequests中，InFlightRequests保存对象的具体形式为 Map<Nodeld, Deque<request>>，它的主要作用是缓存了已经发出去但还没有收到服务端响应的请求（Nodeld 是一个 String 类型，表示节点的 id 编号）。与此同时，InFlightRequests 还提供了许多管理类的方法，并且通过配置参数还可以限制每个连接（也就是客户端与 Node之间的连接）最多缓存的请求数。这个配置参数为 max.in.flight.request.per. connection ，默认值为5，即每个连接最多只能缓存5个未响应的请求，超过该数值之后就不能再向这个连接发送更多的请求了，除非有缓存的请求收到了响应（ Response ）。通过比较 Deque<Request> 的size与这个参数的大小来判断对应的 Node中是否己经堆积了很多未响应的消息，如果真是如此，那么说明这个 Node 节点负载较大或网络连接有问题，再继续发送请求会增大请求超时的可能。

6.3.1Producer往Broker发送消息应答机制

kafka 在 producer 里面提供了消息确认机制。我们可以通过配置来决定消息发送到对应分区的几个副本才算消息发送成功。可以在构造producer 时通过acks参数指定（在 0.8.2.X 前是通过 request.required.acks 参数设置的）。这个参数支持以下三种值：

acks = 0：意味着如果生产者能够通过网络把消息发送出去，那么就认为消息已成功写入 kafka 。在这种情况下还是有可能发生错误，比如发送的对象不能被序列化或者网卡发生故障，但如果是分区离线或整个集群长时间不可用，那就不会收到任何错误。在 acks=0 模式下的运行速度是非常快的（这就是为什么很多基准测试都是基于这个模式），你可以得到惊人的吞吐量和带宽利用率，不过如果选择了这种模式，大概率会丢失一些消息。

acks = 1：意味着leader 在收到消息并把它写入到分区数据文件（不一定同步到磁盘上）时会返回确认或错误响应。在这个模式下，如果发生正常的 leader 选举，生产者会在选举时收到一个 LeaderNotAvailableException 异常，如果生产者能恰当地处理这个错误，它会重试发送悄息，最终消息会安全到达新的 leader 那里。不过在这个模式下仍然有可能丢失数据，比如消息已经成功写入 leader，但在消息被复制到 follower 副本之前 leader发生崩溃。

acks = all（这个和 request.required.acks = -1 含义一样）：意味着 leader 在返回确认或错误响应之前，会等待所有同步副本都收到悄息。如果和 min.insync.replicas 参数结合起来，就可以决定在返回确认前至少有多少个副本能够收到悄息，生产者会一直重试直到消息被成功提交。不过这也是最慢的做法，因为生产者在继续发送其他消息之前需要等待所有副本都收到当前的消息。

根据实际的应用场景，我们设置不同的 acks，以此保证数据的可靠性。

acks	含义
0	Producer往集群发送数据不需要等到集群的确认信息，不确保消息发送成功。安全性最低但是效率最高。
1	Producer往集群发送数据只要 leader成功写入消息就可以发送下一条，只确保Leader 接收成功。
-1或all	Producer往集群发送数据需要所有的ISR Follower 都完成从 Leader 的同步才会发送下一条，确保 Leader发送成功和所有的副本都成功接收。安全性最高，但是效率最低。

生产者将acks设置为all，是否就一定不会丢数据呢？

否！如果在某个时刻ISR列表只剩leader自己了，那么就算acks=all，收到这条数据还是只有一个点；

可以配合另外一个参数缓解此情况：最小同步副本数>=2

BROKER端参数： min.insync.replicas（默认1）

生产者的ack=all，也不能完全保证数据发送的100%可靠性

为什么？因为，如果服务端目标partition的同步副本只有leader自己了，此时，它收到数据就会给生产者反馈成功！

可以修改服务端的一个参数（分区最小ISR数[min.insync.replicas]>=2），来避免此问题；

6.3.2其他的生产者参数

acks

acks是控制kafka服务端向生产者应答消息写入成功的条件；生产者根据得到的确认信息，来判断消息发送是否成功；

max.request.size

这个参数用来限制生产者客户端能发送的消息的最大值，默认值为 1048576B ，即 1MB

一般情况下，这个默认值就可以满足大多数的应用场景了。

这个参数还涉及一些其它参数的联动，比如 broker 端（topic级别参数）的 message.max.bytes参数（默认1000012），如果配置错误可能会引起一些不必要的异常；比如将 broker 端的 message.max.bytes 参数配置为10B ，而 max.request.size参数配置为20B，那么当发送一条大小为 15B 的消息时，生产者客户端就会报出异常；

retries和retry.backoff.ms ==> 间隔时间 避免无效的重试

retries参数用来配置生产者重试的次数，默认值为2147483647，即在发生异常的时候进行任何重试动作。

消息在从生产者发出到成功写入服务器之前可能发生一些临时性的异常，比如网络抖动、 leader 副本的选举等，这种异常往往是可以自行恢复的，生产者可以通过配置 retries大于0的值，以此通过内部重试来恢复而不是一味地将异常抛给生产者的应用程序。如果重试达到设定的次数，那么生产者就会放弃重试并返回异常。重试还和另一个参数 retry.backoff.ms 有关，这个参数的默认值为100，它用来设定两次重试之间的时间间隔，避免无效的频繁重试。如果将 retries参数配置为非零值，并且 max .in.flight.requests.per.connection 参数配置为大于1的值，那可能会出现错序的现象：如果批次1消息写入失败，而批次2消息写入成功，那么生产者会重试发送批次1的消息，此时如果批次1的消息写入成功，那么这两个批次的消息就出现了错序。

对于某些应用来说，顺序性非常重要，比如MySQL binlog的传输，如果出现错误就会造成非常严重的后果；一般而言，在需要保证消息顺序的场合建议把参数max.in.flight.requests.per.connection 配置为1 ，而不是把retries配置为0，不过这样也会影响整体的吞吐。

compression.type

这个参数用来指定消息的压缩方式，默认值为“none"，即默认情况下，消息不会被压缩。该参数还可以配置为 "gzip"，"snappy" 和 "lz4"。对消息进行压缩可以极大地减少网络传输、降低网络I/O，从而提高整体的性能。消息压缩是一种以时间换空间的优化方式，如果对时延有一定的要求，则不推荐对消息进行压缩；

batch.size

每个Batch要存放batch.size大小的数据后，才可以发送出去。比如说batch.size默认值是16KB，那么里面凑够16KB的数据才会发送。理论上来说，提升batch.size的大小，可以允许更多的数据缓冲在recordAccumulator里面，那么一次Request发送出去的数据量就更多了，这样吞吐量可能会有所提升。但是batch.size也不能过大，要是数据老是缓冲在Batch里迟迟不发送出去，那么发送消息的延迟就会很高。一般可以尝试把这个参数调节大些，利用生产环境发消息负载测试一下。

linger.ms

这个参数用来指定生产者发送 ProducerBatch 之前等待更多消息（ ProducerRecord ）加入

ProducerBatch 时间，默认值为0。生产者客户端会在ProducerBatch填满或等待时间超过linger.ms 值时发送出去。

增大这个参数的值会增加消息的延迟，但是同时能提升一定的吞吐量。

enable.idempotence

是否开启幂等性功能，详见后续原理加强；

幂等性，就是一个操作重复做，也不会影响最终的结果！

int a = 1;

a++; // 非幂等操作

val map = new HashMap()

map.put(“a”,1); // 幂等操作

在kafka中，同一条消息，生产者如果多次重试发送，在服务器中的结果如果还是只有一条，这就是具备幂等性；否则，就不具备幂等性！

partitioner.class

用来指定分区器，默认：org.apache.kafka.internals.DefaultPartitioner

默认分区器的分区规则：

如果数据中有key，则按key的murmur hash值 % topic分区总数得到目标分区

如果数据只有value，则在各个分区间轮询（老版本，新版本是new出来的一个随机数）

自定义partitioner需要实现org.apache.kafka.clients.producer.Partitioner接口

练一练：如何实现精准一次性消费

将kafka中的数据读出来，写入到mysql中

Java
package com.doit.day01;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.sql.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

public class KafkaToMysql {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"doit03");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456");
        conn.setAutoCommit(false);

        PreparedStatement pps = conn.prepareStatement("insert into user_info values(?,?,?)");
        PreparedStatement pps_off = conn.prepareStatement("insert into t_offset values(?,?) on DUPLICATE key UPDATE offset = ?");
        PreparedStatement off = conn.prepareStatement("select offset from t_offset where topic_partition = ?");

//        consumer.subscribe(Arrays.asList("k2m1"));
        consumer.subscribe(Arrays.asList("k2m1"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            }
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                try    {
                for (TopicPartition partition : partitions) {
                    int partition1 = partition.partition();
                    String topic = partition.topic();

                        off.setString(1,topic+"_"+partition1);
                        ResultSet resultSet = off.executeQuery();
                        while (resultSet.next()){
                            long offset = resultSet.getLong(1);
                            System.out.println("发生了消费者再均衡了，分区啥的都重新分配了,新的方案是："+topic+","+partition1);
                            consumer.seek(new TopicPartition(topic,partition1),offset);
                        }
                    }

                }catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        });

        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMinutes(Integer.MAX_VALUE));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    String value = record.value();
                    String[] arr = value.split(",");
                    pps.setInt(1,Integer.parseInt(arr[0]));
                    pps.setString(2,arr[1]);
                    pps.setString(3,arr[2]);
                    pps.execute();

                    String topic = record.topic();
                    int partition = record.partition();
                    long offset = record.offset();
                    pps_off.setString(1,topic+"_"+partition);
                    pps_off.setLong(2,offset+1);
                    pps_off.setLong(3,offset+1);

//                    if (arr[0].equals("4")){
//                        throw new Exception("自己造了个一场抛一下");
//                    }
                    pps_off.execute();

                    //提交事务
                    conn.commit();
                } catch (Exception e) {
                    e.printStackTrace();
                    //有异常了我就回滚事务
                    conn.rollback();
                }
            }
        }
    }
}

6.4消费者组再均衡分区分配策略

会触发rebalance（消费者）的事件可能是如下任意一种：

有新的消费者加入消费组。

有消费者宕机下线，消费者并不一定需要真正下线，例如遇到长时间的 GC 、网络延迟导致消费者长时间未向GroupCoordinator发送心跳等情况时，GroupCoordinator 会认为消费者己下线。

有消费者主动退出消费组（发送LeaveGroupRequest 请求）：比如客户端调用了unsubscrible()方法取消对某些主题的订阅。

消费组所对应的 GroupCoorinator节点发生了变更。

消费组内所订阅的任一主题或者主题的分区数量发生变化。

将分区的消费权从一个消费者移到另一个消费者称为再均衡（rebalance），如何rebalance也涉及到分区分配策略。

kafka有两种的分区分配策略：range（默认）和 roundrobin（新版本中又新增了另外2种）

我们可以通过partition.assignment.strategy参数选择 range 或 roundrobin。

partition.assignment.strategy参数默认的值是range。

partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

partition.assignment.strategy=org.apache.kafka.clients.consumer.RangeAssignor

6.4.1Range Strategy

先将消费者按照client.id字典排序，然后按topic逐个处理；

针对一个topic，将其partition总数/消费者数得到商n和余数m，则每个consumer至少分到n个分区，且前m个consumer每人多分一个分区；

举例说明1：假设有TOPIC_A有5个分区，由3个consumer（C1,C2,C3）来消费；5/3得到商1，余2，则每个消费者至少分1个分区，前两个消费者各多1个分区C1: 2个分区，C2:2个分区,C3:1个分区

接下来，就按照“区间”进行分配：

TOPIC_A-0 TOPIC_A-1 TOPIC_A-2 TOPIC_A_3 TOPIC_A-4

C1: TOPIC_A-0 TOPIC_A-1

C2 : TOPIC_A-2 TOPIC_A_3

C3: TOPIC_A-4

举例说明2：假设TOPIC_A有5个分区，TOPIC_B有3个分区，由2个consumer（C1,C2）来消费

先分配TOPIC_A：

5/2得到商2，余1，则C1有3个分区，C2有2个分区，得到结果

C1: TOPIC_A-0 TOPIC_A-1 TOPIC_A-2

C2: TOPIC_A-3 TOPIC_A-4

再分配TOPIC_B

3/2得到商1，余1，则C1有2个分区，C2有1个分区，得到结果

C1: TOPIC_B-0 TOPIC_B-1

C2: TOPIC_B-2

最终分配结果：

C1: TOPIC_A-0 TOPIC_A-1 TOPIC_A-2 TOPIC_B-0 TOPIC_B-1

C2: TOPIC_A-3 TOPIC_A-4 TOPIC_B-2

6.4.2Round-Robin Strategy

将所有主题分区组成TopicAndPartition列表，并对TopicAndPartition列表按照其hashCode 排序

然后，以轮询的方式分配给各消费者

以上述“例2”来举例：

先对TopicPartition的hashCode排序，假如排序结果如下：

TOPIC_A-0 TOPIC_B-0 TOPIC_A-1 TOPIC_A-2 TOPIC_B-1 TOPIC_A-3 TOPIC_A-4 TOPIC_B-2

然后按轮询方式分配

C1: TOPIC_A-0 TOPIC_A-1 TOPIC_B-1

C2: TOPIC_B-0 TOPIC_A-2 TOPIC_A-3

C3 TOPIC_A-4

6.4.3Sticky Strategy

对应的类叫做： org.apache.kafka.clients.consumer.StickyAssignor

sticky策略的特点：

要去达成最大化的均衡

尽可能保留各消费者原来分配的分区

再均衡的过程中，还是会让各消费者先取消自身的分区，然后再重新分配（只不过是分配过程中会尽量让原来属于谁的分区依然分配给谁）

6.4.4Cooperative Sticky Strategy

对应的类叫做： org.apache.kafka.clients.consumer.ConsumerPartitionAssignor

sticky策略的特点：

逻辑与sticky策略一致

支持cooperative再均衡机制（再均衡的过程中，不会让所有消费者取消掉所有分区然后再进行重分配）

6.5消费者组再均衡流程

消费组在消费数据的时候，有两个角色进行组内的各事务的协调；

角色1： Group Coordinator （组协调器）位于服务端（就是某个broker）

组协调器的定位：

Plain Text
coordinator在我们组记偏移量的__consumer_offsets分区的leader所在broker上
查找Group Coordinator的方式：
先根据消费组groupid的hashcode值计算它应该所在__consumer_offsets 中的分区编号；分区数
Utils.abs(groupId.hashCode) % groupMetadataTopicPartitionCount
groupMetadataTopicPartitionCount为__consumer_offsets的分区总数，这个可以通过broker端参数offset.topic.num.partitions来配置，默认值是50；
找到对应的分区号后，再寻找此分区leader副本所在broker节点，则此节点即为自己的Grouping Coordinator；

角色2： Group Leader （组长）位于消费端（就是消费组中的某个消费者）

组长的定位：随机选的哦！！！

6.5.1GroupCoordinator介绍

每个消费组在服务端对应一个GroupCoordinator其进行管理，GroupCoordinator是Kafka服务端中用于管理消费组的组件。

消费者客户端中由ConsumerCoordinator组件负责与GroupCoordinator行交互；

ConsumerCoordinator和GroupCoordinator最重要的职责就是负责执行消费者rebalance操作

6.5.2再均衡流程

eager协议的再均衡过程整体流程如下图：

特点：再均衡发生时，所有消费者都会停止工作，等待新方案的同步

Cooperative协议的再均衡过程整体流程如下图：

特点：cooperative把原来eager协议的一次性全局再均衡，化解成了多次的小均衡，并最终达到全局均衡的收敛状态

6.5.3再均衡监听器

如果想控制消费者在发生再均衡时执行一些特定的工作，可以通过订阅主题时注册“再均衡监听器”来实现；

场景举例：在发生再均衡时，处理消费位移

如果A消费者消费掉的一批消息还没来得及提交offset，而它所负责的分区在rebalance中转移给了B消费者，则有可能发生数据的重复消费处理。此情形下，可以通过再均衡监听器做一定程度的补救；

代码示例：

Java
package com.doitedu;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Optional;
import java.util.Properties;

/**
* 消费组再均衡观察
*/

public class ConsumerDemo2 {
    public static void main(String[] args) {
        //1.创建kafka的消费者对象，附带着把配置文件搞定
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"g01");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        //2.订阅主题(确定需要消费哪一个或者多个主题)
        //我现在想看看如果我的消费者组里面，多了一个消费者或者少了一个消费者，他有没有给我做再均衡
        consumer.subscribe(Arrays.asList("reb-1", "reb-2"), new ConsumerRebalanceListener() {
            /**
             * 这个方法是将原来的分配情况全部取消，或者说把所有的分区全部回收了
             * 这个全部取消很恶心，原来的消费者消费的好好的，他一下子就给他全部停掉了
             * @param collection
             */
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                System.out.println("我原来的均衡情况是："+collection + "我已经被回收了！！");
            }
            /**
             * 这个方法是当上面的分配情况全部取消以后，调用这个方法，来再次分配，这是在均衡分配后的情况
             * @param collection
             */
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                System.out.println("我是重新分配后的结果："+collection);
            }
        });

        while (true){
            consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
        }

    }
}

6.6kafka系统的CAP保证

CAP理论作为分布式系统的基础理论,它描述的是一个分布式系统在以下三个特性中：

一致性（Consistency）

可用性（Availability)

分区容错性（Partition tolerance）

最多满足其中的两个特性。也就是下图所描述的。分布式系统要么满足CA，要么CP，要么AP。无法同时满足CAP。

分区容错性：指的分布式系统中的某个节点或者网络分区出现了故障的时候，整个系统仍然能对外提供满足一致性和可用性的服务。也就是说部分故障不影响整体使用。事实上我们在设计分布式系统时都会考虑到bug,硬件，网络等各种原因造成的故障，所以即使部分节点或者网络出现故障，我们要求整个系统还是要继续使用的(不继续使用,相当于只有一个分区,那么也就没有后续的一致性和可用性了)

可用性：一直可以正常的做读写操作。简单而言就是客户端一直可以正常访问并得到系统的正常响应。用户角度来看就是不会出现系统操作失败或者访问超时等问题。

一致性：在分布式系统完成某写操作后任何读操作，都应该获取到该写操作写入的那个最新的值。相当于要求分布式系统中的各节点时时刻刻保持数据的一致性。

Kafka 作为一个商业级消息中间件，数据可靠性和可用性是优先考虑的重点，兼顾数据一致性；

参考文档：https://www.cnblogs.com/lilpig/p/16840963.html

6.6.1分区副本机制

kafka 从 0.8.0 版本开始引入了分区副本；引入了数据冗余

用CAP理论来说，就是通过副本及副本leader动态选举机制提高了kafka的 分区容错性和可用性

但从而也带来了数据一致性的巨大困难！

6.6.2分区副本的数据一致性困难

kafka让分区多副本同步的基本手段是： follower副本定期向leader请求数据同步！

既然是定期同步，则leader和follower之间必然存在各种数据不一致的情景！

问题1：分区副本间动态不一致

问题2：消费者所见不一致

如果此时leader宕机，follower1或follower2被选为新的leader，则leader换届前后，消费者所能读取到的数据发生了不一致；

问题3：分区副本间最终不一致

6.6.2一致性问题解决方案（HW）

动态过程中的副本数据不一致，是很难解决的；

kafka先尝试着解决上述“消费者所见不一致”及“副本间数据最终不一致”的问题；

解决方案的核心思想

在动态不一致的过程中，维护一条步进式的“临时一致线”(既所谓的High Watermark)；

高水位线HW = ISR副本中最小LEO(副本的最大消息位移+1)；

底层逻辑就是：offset<HW的message，是各副本间一致的且安全的！

解决“消费者所见不一致” （消费者只允许看到HW以下的message）

解决“分区副本数据最终不一致” （follower数据按HW截断）

6.6.3HW方案的天生缺陷

如前所述，看似HW解决了“分区数据最终不一致”的问题，以及“消费者所见不一致”的问题，但其实，这里面存在一个巨大的隐患，导致：

“分区数据最终不一致”的问题依然存在

producer设置acks=all后，依然有可能丢失数据的问题

产生如上结果的根源是：HW高水位线的更新，与数据同步的进度，存在迟滞！

第一次fetch请求，分leader端和follower端：

leader端：

读取底层log数据。

根据fetch带过来的offset=0的数据（就是follower的LEO，因为follower还没有写入数据，因此LEO=0），更新remote LEO为0。

一轮结束后尝试更新HW，做min(leader LEO,remote LEO)的计算，结果为0。

把读取到的三条log数据，加上leader HW=0，一起发给follower副本。

follower端：

写入数据到log文件，更新自己的LEO=3。

更新HW，做min(leader HW,follower LEO)的计算，由于leader HW=0，因此更新后HW=0。

可以看出，第一次fetch请求后，leader和follower都成功写入了三条消息，但是HW都依然是0，对消费者来说都是不可见的，还需要第二次fetch请求。

第二次fetch请求，分leader端和follower端：

leader端：

读取底层log数据。

根据fetch带过来的offset=3的数据（上一次请求写入了数据，因此LEO=3），更新remote LEO为3。

尝试更新HW，做min(leader LEO,remote LEO)的计算，结果为3。

把读取到的log数据（其实没有数据），加上leader HW=3，一起发给follower副本。

follower端：

写入数据到log文件，没有数据可以写，LEO依然是3。

更新HW，做min(leader HW,follower LEO)的计算，由于leader HW=3，因此更新后HW=3。

这个时候，才完成数据的写入，并且分区HW（分区HW指的就是leader副本的HW）更新为3，代表消费者可以消费offset=0,1,2的三条消息了，上面的过程就是kafka处理消息写入和备份的全流程。

从以上步骤可看出，leader 中保存的 remote LEO 值的更新（也即HW的更新）总是需要额外一轮 fetch RPC 请求才能完成，这意味着在 leader 切换过程中，会存在数据丢失以及数据不一致的问题！

6.6.4HW会产生数据丢失和副本最终不一致问题

数据丢失的问题（即使produce设置acks=all，依然会发生）

如上图所示：

状态起始：最新消息c已同步，但是水位线还没开始同步

在此时leader崩溃（即 follower 没能通过下一轮请求来更新 HW 值）

follower成为了leader，会自动将 LEO 值调整到之前的 HW 值，即会进行日志截断

然后，原来的leader重启上线，会向新的leader发送请求请求，收到 fetch 响应后，拿到 HW 值，并更新本地 HW 值，发现我也要截取，悲剧发生了，数据丢了

副本间数据最终不一致的问题（即使produce设置acks=all，依然会发生）

如上图所示：

状态起始：最新消息c已同步，但是水位线还没开始同步

在此时leader崩溃（即 follower 没能通过下一轮请求来更新 HW 值）

follower成为了leader，会自动将 LEO 值调整到之前的 HW 值，即会进行日志截断

在截断日志之后，也就是这个d被截断了之后，我又加了一条数据是e

然后，原来的leader重启上线，会向新的leader发送请求请求，收到 fetch 响应后，拿到 HW 值，并更新本地 HW 值，发现我的数据和leader的数据一样，好的，我就不用截取了，我更新HW就好了，就这样，一个新的悲剧又发生了，数据不一致了

只要新一届leader在老leader重启上线前，接收了新的数据，就可能发生上图中的场景，根源也在于HW的更新落后于数据同步进度

6.6.5Leader-Epoch机制的引入

为了解决 HW 更新时机是异步延迟的，而 HW 又是决定日志是否备份成功的标志，从而造成数据丢失和数据不一致的现象，Kafka 引入了 leader epoch 机制；

在每个副本日志目录下都创建一个 leader-epoch-checkpoint 文件，用于保存 leader 的 epoch 信息；

leader-epoch的含义

如下，leader epoch 长这样：

它的格式为 (epoch offset)，epoch指的是 leader 版本，它是一个单调递增的一个正整数值，每次 leader 变更，epoch 版本都会 +1，offset 是每一代 leader 写入的第一条消息的位移值，比如：

(0,0)

(1,300)

以上第2个版本是从位移300开始写入消息，意味着第一个版本写入了 0-299 的消息。

leader epoch 具体的工作机制

当副本成为 leader 时：

这时，如果此时生产者有新消息发送过来，会首先更新leader epoch 以及LEO ，并添加到 leader-epoch-checkpoint 文件中；

当副本变成 follower 时：

发送LeaderEpochRequest请求给leader副本，该请求包括了follower中最新的epoch 版本；

leader返回给follower的响应中包含了一个LastOffset，如果 follower last epoch = leader last epoch（纪元相同），则 LastOffset = leader LEO，否则取follower last epoch 中最小的 leader epoch 的 start offset 值；

举个例子：假设 follower last epoch = 1，此时 leader 有 (1, 20) (2, 80) (3, 120)，则 LastOffset = 80；

follwer 拿到 LastOffset 之后，会对比当前 LEO 值是否大于 LastOffset，如果当前 LEO 大于 LastOffset，则从 LastOffset 截断日志；

follower 开始发送 fetch 请求给 leader 保持消息同步。

leader epoch 如何解决HW的备份缺陷

解决数据丢失和数据不一致的问题

如上图所示：

follower当选leader后，收到纪元消息，发现 LastOffset等于当前 LEO 值，故不用进行日志截断。

follower重启后同步消息，发现自己也不用截取，数据一致，齐活儿

当然，如果说后来增加消息以后，也不需要截取，直接同步数据就行(当ack=-1)

6.6.6LEO/HW/LSO等相关术语速查

LEO:（last end offset）就是该副本中消息的最大偏移量的值+1 ；

HW:（high watermark）各副本中LEO的最小值。这个值规定了消费者仅能消费HW之前的数据；

LW：（low watermark）一个副本的log中，最小的消息偏移量；应该是和log里面的偏移量有关系

LSO：（last stable offset）最后一个稳定的offset；对未完成的事务而言，LSO 的值等于事务中第一条消息的位置(firstUnstableOffset)，对已完成的事务而言，它的值同 HW 相同；

LEO与HW 与数据一致性密切相关；

如图，各副本中最小的LEO是3，所以HW是3，所以，消费者此刻最多能读到Msg2;

6.6.7不清洁选举[了解]

不清洁选举，是指允许“非ISR副本”可以被选举为leader；非ISR副本被选举为leader，将极大增加数据丢失及数据不一致的可能性！由参数 unclean.leader.election.enable=false（默认）控制；

初始状态： follower2严重落后于leader，并且不属于ISR副本

此刻，所有ISR副本宕机：

Follower2成为新的leader，并接收数据

之前宕机副本重启，按照最新leader的最新leo进行截断，产生数据丢失及不一致

6.7.幂等性

6.7.1幂等性要点

Kafka 0.11.0.0 版本开始引入了幂等性与事务这两个特性，以此来实现 EOS ( exactly once

semantics ，精确一次处理语义）

生产者在进行发送失败后的重试时（retries），有可能会重复写入消息，而使用 Kafka幂等性功能之后就可以避免这种情况。

开启幂等性功能，只需要显式地将生产者参数 enable.idempotence设置为 true （默认值为 false）：

props.put("enable.idempotence",true);

在开启幂等性功能时，如下几个参数必须正确配置：

retries > 0

max.in.flight.requests.per.connection<=5

acks = -1

如有违反，则会抛出ConfigException异常；

6.7.2kafka幂等性实现机制

1）每一个producer在初始化时会生成一个producer_id，并为每个目标分区维护一个“消息序列号”；

2）producer每发送一条消息，会将<producer_id,分区>对应的“序列号”加1

3）broker端会为每一对{producer_id,分区}维护一个序列号，对于每收到的一条消息，会判断服务端的SN_OLD和接收到的消息中的SN_NEW进行对比：

如果SN_OLD + 1 == SN_NEW，正常；

如果SN_NEW<SN_OLD+1，说明是重复写入的数据，直接丢弃

如果SN_NEW>SN_OLD+1，说明中间有数据尚未写入，或者是发生了乱序，或者是数据丢失，将抛出严重异常：OutOfOrderSequenceException

producer.send(“aaa”) 消息aaa就拥有了一个唯一的序列号

如果这条消息发送失败，producer内部自动重试（retry），此时序列号不变；

producer.send(“bbb”) 消息bbb拥有一个新的序列号

注意：kafka只保证producer单个会话中的单个分区幂等；

6.8.kafka事务(伪事务)

6.8.1事务要点知识

Kafka的事务控制原理

主要原理：开始事务-->发送一个ControlBatch消息（事务开始）

提交事务-->发送一个ControlBatch消息（事务提交）

放弃事务-->发送一个ControlBatch消息（事务终止）

开启事务的必须配置参数（我不支持数据得回滚，但是我能做到，一荣俱荣，一损俱损）

Java
Properties props = new Properties();
props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"doit01:9092");
props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// acks
props.setProperty(ProducerConfig.ACKS_CONFIG,"-1");
// 生产者的重试次数
props.setProperty(ProducerConfig.RETRIES_CONFIG,"3");
// 飞行中的请求缓存最大数量
props.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,"3");
// 开启幂等性
props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,"true");
// 设置事务id
props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"trans_001");

事务控制的代码模板

Java
// 初始化事务
producer.initTransaction( )

// 开启事务
producer.beginTransaction( )

// 干活

// 提交事务
producer.commitTransaction( )

// 异常回滚（放弃事务） catch里面
producer.abortTransaction( )

消费者api是会拉取到尚未提交事务的数据的；只不过可以选择是否让用户看到！

是否让用户看到未提交事务的数据，可以通过消费者参数来配置：

isolation.level=read_uncommitted（默认值）

isolation.level=read_committed

kafka还有一个“高级”事务控制，只针对一种场景：

用户的程序，要从kafka读取源数据，数据处理的结果又要写入kafka

kafka能实现端到端的事务控制（比起上面的“基础”事务，多了一个功能，通过producer可以将consumer的消费偏移量绑定到事务上提交）

Java
producer.sendOffsetsToTransaction(offsets,consumer_id)

6.8.2事务api示例

为了实现事务，应用程序必须提供唯一transactional.id，并且开启生产者的幂等性

Java
properties.put ("transactional.id","transactionid00001");
properties.put ("enable.idempotence",true);

kafka生产者中提供的关于事务的方法如下：

“消费kafka-处理-生产结果到kafka”典型场景下的代码结构示例：

Java
package com.doit.day04;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class Exercise_kafka2kafka {
    public static void main(String[] args) {

        Properties props = new Properties();
        //消费者的
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "shouwei");
        //自动提交偏移量
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

        //写生产者的一些属性
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //设置ack 开启幂等性必须设置的三个参数
        props.setProperty(ProducerConfig.ACKS_CONFIG,"-1");
        props.setProperty(ProducerConfig.RETRIES_CONFIG,"3");
        props.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,"3");
        //开启幂等性
        props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,"true");
        //开启事务
        props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"doit40");

        //消费数据
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        //初始化事务
        producer.initTransactions();
        //订阅主题
        consumer.subscribe(Arrays.asList("eventlog"));
        while (true){
            //拉取数据
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
            try {
                //开启事务
                producer.beginTransaction();
                for (ConsumerRecord<String, String> record : poll) {
                    String value = record.value();
                    //将value的值写入到另外一个topic中
                    producer.send(new ProducerRecord<String,String>("k2k",value));
                }
                producer.flush();
                //提交偏移量
                consumer.commitAsync();
                //提交事务
                producer.commitTransaction();

            } catch (ProducerFencedException e) {
                //放弃事务
                producer.abortTransaction();
            }
        }
    }
}

6.8.3事务实战案例

在实际数据处理中，consume-transform-produce是一种常见且典型的场景；

在此场景中，我们往往需要实现，从“读取source数据，至业务处理，至处理结果写入kafka”的整个流程，具备原子性：

要么全流程成功，要么全部失败！

（处理且输出结果成功，才提交消费端偏移量；处理或输出结果失败，则消费偏移量也不会提交）

要实现上述的需求，可以利用Kafka中的事务机制：

它可以使应用程序将消费消息、生产消息、提交消费位移当作原子操作来处理，即使该生产或消费会跨多个topic分区；

在消费端有一个参数isolation.level，与事务有着莫大的关联，这个参数的默认值为“read_uncommitted”，意思是说消费端应用可以看到（消费到）未提交的事务，当然对于已提交的事务也是可见的。这个参数还可以设置为“read_committed”，表示消费端应用不可以看到尚未提交的事务内的消息。

控制消息（ControlBatch：COMMIT/ABORT）表征事务是被提交还是被放弃

6.9分区数与吞吐量

Kafka本身提供用于生产者性能测试的kafka-producer-perf-test.sh 和用于消费者性能测试的 kafka-consumer-perf-test. sh，主要参数如下：

topic 用来指定生产者发送消息的目标主题；

num-records 用来指定发送消息的总条数

record-size 用来设置每条消息的字节数；

producer-props 参数用来指定生产者的配置，可同时指定多组配置，各组配置之间以空格分隔与 producer-props 参数对应的还有一个 producer-config参数，它用来指定生产者的配置文件；

throughput 用来进行限流控制，当设定的值小于0时不限流，当设定的值大于0时，当发送的吞吐量大于该值时就会被阻塞一段时间。

经验：如何把kafka服务器的性能利用到最高，一般是让一台机器承载（ cpu线程数*2~3 ）个分区

测试环境：节点3个，cpu 2核2线程，内存8G ，每条消息1k

测试结果： topic在12个分区时，写入、读取的效率都是达到最高

写入： 75MB/s ，7.5万条/s

读出： 310MB/s ，31万条/s

当分区数>12 或者 <12 时，效率都比=12时要低！

6.10性能测试

6.10.1生产者性能测试

tpc_3： 分区数3，副本数1

Shell
[root@doitedu01 kafka_2.11-2.0.0]# bin/kafka-producer-perf-test.sh --topic tpc_3 --num-records 100000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=linux01:9092 acks=1

100000 records sent, 26723.677178 records/sec (26.10 MB/sec), 818.30 ms avg latency, 1689.00 ms max latency, 595 ms 50th, 1580 ms 95th, 1649 ms 99th, 1687 ms 99.9th.

tpc_4： 分区数4，副本数2

Shell
[root@doitedu01 kafka_2.11-2.0.0]# bin/kafka-producer-perf-test.sh --topic tpc_4 --num-records 100000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=doitedu01:9092 acks=1

100000 records sent, 25886.616619 records/sec (25.28 MB/sec), 962.06 ms avg latency, 1647.00 ms max latency, 857 ms 50th, 1545 ms 95th, 1622 ms 99th, 1645 ms 99.9th.

tpc_5：分区数5，副本数1

Shell
[root@doitedu01 kafka_2.11-2.0.0]# bin/kafka-producer-perf-test.sh --topic tpc_5 --num-records 100000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=doitedu01:9092 acks=1

100000 records sent, 28785.261946 records/sec (28.11 MB/sec), 789.29 ms avg latency, 1572.00 ms max latency, 665 ms 50th, 1502 ms 95th, 1549 ms 99th, 1564 ms 99.9th.

tpc_6：分区数6，副本数1

Shell
[root@doitedu01 kafka_2.11-2.0.0]# bin/kafka-producer-perf-test.sh --topic tpc_6 --num-records 100000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=doitedu01:9092 acks=1

100000 records sent, 42662.116041 records/sec (41.66 MB/sec), 508.68 ms avg latency, 1041.00 ms max latency, 451 ms 50th, 945 ms 95th, 1014 ms 99th, 1033 ms 99.9th.

tpc_12：分区数12

Shell
[root@doitedu01 kafka_2.11-2.0.0]# bin/kafka-producer-perf-test.sh --topic tpc_12 --num-records 100000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=doitedu01:9092 acks=1

100000 records sent, 56561.085973 records/sec (55.24 MB/sec), 371.42 ms avg latency, 1103.00 ms max latency, 314 ms 50th, 988 ms 95th, 1091 ms 99th, 1093 ms 99.9th.

脚本还包含了许多其他的参数，比如 from latest group、print-metrics、threads等，篇幅及时间限制，同学们可以自行了解这些参数的使用细节。

例如，加上参数： --print-metrics，则会打印更多信息

6.10.2消费者性能测试

Shell
[root@doitedu01 kafka_2.11-2.0.0]# bin/kafka-consumer-perf-test.sh --topic tpc_3 --messages 100000 --broker-list doitedu01:9092 --consumer.config x.properties

结果数据个字段含义：

start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec

2020-11-14 15:43:42:422, 2020-11-14 15:43:43:347, 98.1377, 106.0948, 100493, 108641.0811, 13, 912, 107.6071, 110189.6930

结果中包含了多项信息，分别对应起始运行时间（start. time）、结束运行时 end.time）、消息总量（data.consumed.in.MB ，单位为 MB ），按字节大小计算的消费吞吐量（单位为 MB ）、消费的消息总数（ data. consumed.in nMsg ）、按消息个数计算的吞吐量（nMsg.sec）、再平衡的时间（ rebalance time.ms 单位为MB/s）、拉取消息的持续时间（fetch.time.ms，单位为ms）、每秒拉取消息的字节大小（fetch.MB.sec 单位 MB/s）、每秒拉取消息的个数( fetch.nM.sec）。其中 fetch.time.ms= end.time - start.time - rebalance.time.ms

6.10.3分区数与吞吐量实际测试

Kafka只允许单个分区中的消息被一个消费者线程消费，一个消费组的消费并行度完全依赖于所消费的分区数；

如此看来，如果一个主题中的分区数越多，理论上所能达到的吞吐量就越大，那么事实真的如预想的一样吗？

我们以一个3台普通阿里云主机组成的3节点kafka集群进行测试，每台主机的内存大小为8GB，磁盘为40GB，4核CPU 16线程，主频2600MHZ，JVM版本为1.8.0_112，Linux系统版本为2.6.32-504.23.4.el6.x86_64。

创建分区数为1、20、50、100、200、500、1000的主题，对应的主题名称分别为 topic-1 topic 20 topic-50 topic-100 topic-200 topic-500 topic-1000 ，所有主题的副本因子都设置为1。

消费者，测试结果与上图趋势类同

如何选择合适的分区数？从某种意恩来说，考验的是决策者的实战经验，更透彻地说，是Kafka本身、业务应用、硬件资源、环境配置等多方面的考量而做出的选择。在设定完分区数，或者更确切地说是创建主题之后，还要对其追踪、监控、调优以求更好地利用它。

一般情况下，根据预估的吞吐量及是否与 key 相关的规则来设定分区数即可，后期可以通过增加分区数、增加 broker 或分区重分配等手段来进行改进。

6.10.4分区数设置的经验参考

如果一定要给一个准则，则建议将分区数设定为集群中broker的倍数，即假定集群中有3个broker 节点，可以设定分区数为3/6/9等，至于倍数的选定可以参考预估的吞吐量。

或者根据机器配置的cpu线程数和磁盘性能来设置最大效率的分区数：= CPU线程数 * 1.5~2倍

不过，如果集群中的broker节点数有很多，比如大几十或上百、上千，那么这种准则也不太适用。

还有一个可供参考的分区数设置算法：

每一个分区的写入速度，大约40M/s

每一个分区的读取速度，大约60M/s

假如，数据源产生数据的速度是（峰值）800M/s ，那么为了保证写入速度，该topic应该设置20个分区（副本因子为3）

6.11Kafka速度快的原因（了解）

Kafka速度快的原因：

消息顺序追加（磁盘顺序读写比内存的随机读写还快）

页缓存等技术（数据交给操作系统的页缓存，并不真正刷入磁盘；而是定期刷入磁盘）

使用Zero-Copy （零拷贝）技术来进一步提升性能；`

扩展阅读：零拷贝

所谓的零拷贝是指将数据直接从磁盘文件复制到网卡设备中，而不需要经由应用程序之手；

零拷贝大大提高了应用程序的性能，减少了内核和用户模式之间的上下文切换；对于Linux系统而言，零拷贝技术依赖于底层的 sendfile( )方法实现；对应于Java 语言，FileChannal.transferTo( )方法的底层实现就是 sendfile( )方法；

非零拷贝示意图

零拷贝示意图

零拷贝技术通过DMA (Direct Memory Access）技术将文件内容复制到内核模式下的 Read Buffer。不过没有数据被复制到 Socke Buffer，只有包含数据的位置和长度的信息的文件描述符被加到 Socket Buffer； DMA引擎直接将数据从内核模式read buffer中传递到网卡设备。

这里数据只经历了2次复制就从磁盘中传送出去了，并且上下文切换也变成了2次。

零拷贝是针对内核模式而言的，数据在内核模式下实现了零拷贝；

7.kafka面试题

1. kafka 都有哪些特点？

高吞吐量，低延迟

可以热扩展

并发度高

具有容错性(挂的只剩1台也能正常跑)

可靠性高

2. 请简述你在哪些场景下会选择 kafka？ kafka的一些应用

日志收集：一个公司可以用kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、HBase、Solr等。

消息系统：解耦和生产者和消费者、缓存消息等。

用户行为跟踪：kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。

运营指标：kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

作为流式处理的数据源：比如spark streaming和 Flink

3. kafka 的设计架构你知道吗？

见第五章

4. kafka 分区的目的？

分区对于kafka集群的好处是：实现负载均衡。

分区对于消费者和生产者来说，可以提高并行度，提高效率。--------提高消费者的并行度---》消费者组

5. kafka 是如何做到消息的有序性？

kafka中的每个 partition 中的消息在写入时都是有序的（不断追加），而且单独一个 partition只能由一个消费者去消费，可以在里面保证消息的顺序性。但是分区之间的消息是不保证有序的。

6. kafka 的高可靠性是怎么实现的？

多副本存储

Producer发送数据时可配置ack=all 并且里面有hw 还有leader-epoch

7. 请谈一谈kafka数据一致性原理

一致性指的是不论在什么情况下，Consumer都能读到一致的数据。

HW 高水位线在0.11版本之前，只用了高水位线来保证，但是这个里面其实是会出现一些问题的，比如数据丢失，即使是ack等于-1的情况下，也可能会丢数据

LEO等

在0.11版本之后，新加了一个角色叫leader的纪元号，根据高水位线和纪元号来处理，再配上ack=-1的时候，基本上就不会丢数据了。。。。

8. kafka 在什么情况下会出现消息丢失？

topic的副本如果只有1个，那么一旦这个副本所在broker服务器宕机，则有可能丢失；

producer往kafka写入数据时，如果确认机制参数acks !=all，也可能会造成数据丢失；

不清洁选举机制如果开启，也可能造成数据丢失（不清洁选举就是说在所有ISR副本全部宕机的情况下，可以让OSR副本成为Leader，而OSR中的数据显然不全；那么，就算之前的Leader重新上线了，也会被进行日志截断）

9. 怎么尽可能保证 kafka 的可靠性

副本数>1

ack=all

min.insync.replicas >= 2

10. 数据传输的语义有几种？

数据传输的语义通常有以下三种级别：

设置消费者里面有enable.auto.commit = true/false

最多一次: 消息不会被重复发送，最多被传输一次，但也有可能一次不传输

最少一次: 消息不会被漏发送，最少被传输一次，但也有可能被重复传输

精确一次（Exactly once）: 不会漏传输也不会重复传输

11. kafka 消费者是否可以消费指定分区的消息？

可以，通过assign的方式指定要消费的topic及分区

如果我是subscribe 可以在在均衡监听器的第二个重写方法中使用

12. kafka 消费者是否从指定偏移量开始消费？

可以，通过seek指定偏移量后再开始消费

13. 客户端操作kafka消息是采用poll模式，还是push模式？

kafka最初考虑的问题是，customer应该从brokes拉取消息还是brokers将消息推送到consumer，也就是pull还是push。在这方面，Kafka遵循了一种大部分消息系统共同的传统的设计：producer将消息推送到broker，consumer从broker拉取消息。

一些消息系统比如Scribe和Apache Flume采用了push模式，将消息推送到下游的consumer。这样做有好处也有坏处：由broker决定消息推送的速率，对于不同消费速率的consumer就不太好处理了。消息系统都致力于让consumer以最大的速率最快速的消费消息，但不幸的是，push模式下，当broker推送的速率远大于consumer消费的速率时，consumer恐怕就要崩溃了。最终Kafka还是选取了传统的pull模式。
pull模式的另外一个好处是consumer可以自主决定是否批量的从broker拉取数据。push模式必须在不知道下游consumer消费能力和消费策略的情况下决定是立即推送每条消息还是缓存之后批量推送。如果为了避免consumer崩溃而采用较低的推送速率，将可能导致一次只推送较少的消息而造成浪费。Pull模式下，consumer就可以根据自己的消费能力去决定这些策略。
pull有个缺点是，如果broker没有可供消费的消息，将导致consumer不断在循环中轮询，直到新消息到达。为了避免这点，Kafka有个参数可以让consumer阻塞直到新消息到达(当然也可以阻塞直到消息的数量达到某个特定的量这样就可以批量拉取）

14. kafka的消息格式有了解吗？

V0版本

V1版本

crc attributes mgic timestamp keylength key valuelength value

15. kafka 高效文件存储设计特点

Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。默认存储时间7天

通过索引信息可以快速定位message和确定response的最大大小。

通过index元数据全部映射到memory，可以避免segment file的IO磁盘操作。

通过索引文件稀疏存储，可以大幅降低index文件元数据占用空间大小

16. kafka创建Topic时如何将分区分配给各Broker

副本因子不能大于 Broker 的个数；

第1个分区（partition_0）的第1个副本放置位置是随机从brokerList选择的；

其他分区的第一个副本放置位置相对于partition_0依次往后移。

如果我们有5个 Broker，5个分区，假设第1个分区放在第四个 Broker 上，那么第2个分区将会放在第五个 Broker 上；第3个分区将会放在第一个 Broker 上；第4个分区将会放在第二个 Broker 上，依次类推；

各分区剩余的副本相对于第一个副本放置位置由一个随机数nextReplicaShift决定；

17. kafka的分区分布策略是怎样的？

分区分布的计算策略如下

副本因子不能大于 Broker 的个数；

第一个分区（编号为0）的第一个副本放置位置是随机从 brokerList 选择的；

其他分区的第一个副本放置位置相对于第0个分区依次往后移。也就是如果我们有5个 Broker，5个分区，假设第1个分区放在第四个 Broker 上，那么第2个分区将会放在第五个 Broker 上；第3个分区将会放在第一个 Broker 上；第4个分区将会放在第二个 Broker 上，依次类推；

剩余副本相对于第1个副本放置位置是由一个随机数nextReplicaShift 决定；

18. kafka分区数可以增加或减少吗？为什么？

kafka允许对topic动态增加分区，但不支持减少分区

Kafka 分区数据不支持减少是由很多原因的，比如减少的分区其数据放到哪里去？是删除，还是保留？删除的话，那么这些没消费的消息不就丢了。如果保留这些消息如何放到其他分区里面？追加到其他分区后面的话那么就破坏了 Kafka 单个分区的有序性。如果要保证删除分区数据插入到其他分区保证有序性，那么实现起来逻辑就会非常复杂。

19. kafka新建的分区会在哪创建存储目录

log.dirs参数，其值是 kafka 数据的存放目录；

这个参数可以配置多个目录，目录之间使用逗号分隔，通常这些目录是分布在不同的磁盘上用于提高读写性能。

如果log.dirs参数只配置了一个目录，那么分配到各个 broker 上的分区肯定只能在这个目录下创建文件夹用于存放数据。

但是如果log.dirs参数配置了多个目录，那么 kafka 会在哪个文件夹中创建分区目录呢？答案是：Kafka 会在含有分区目录最少的文件夹中创建新的分区目录，分区目录名为 Topic名+分区ID。

注意，是分区文件夹总数最少的目录，而不是磁盘使用量最少的目录！也就是说，如果你给 log.dirs 参数新增了一个新的磁盘，新的分区目录肯定是先在这个新的磁盘上创建直到这个新的磁盘目录拥有的分区目录不是最少为止。

20. 消费者和消费者组有什么关系？

每个消费者从属于消费组。消费者通过一个参数：group.id 来指定所属的组；

可以把多个消费者的group.id设置成同一个值，那么这几个消费者就属于同一个组；

比如，让c-1，c-2,c-3的group.id=“g1",那么c-1,c-2,c-3这3个消费者都属于g1消费组；

一个消费者，在本质上究竟如何定义：一个消费者可以是一个线程，也可以是一个进程，本质上就是一个consumer对象实例！

消费者组的意义：（可以让多个消费者组成一个组，并共同协作来消费数据，提高消费并行度）一个消费组中的各消费者，在消费一个topic的数据时，互相不重复！如果topic的某分区被组中的一个消费消费，那么，其他消费者就不会再消费这个分区了.

21. 谈一谈 kafka 的消费者组分区分配再均衡

在Kafka中，当有新消费者加入或者订阅的topic数发生变化时，会触发rebalance(再均衡：在同一个消费者组当中，分区的所有权从一个消费者转移到另外一个消费者)机制，Rebalance顾名思义就是重新均衡消费者消费。

Rebalance的过程如下：

第一步：所有成员都向coordinator发送请求，请求入组。一旦所有成员都发送了请求，coordinator会从中选择一个consumer担任leader的角色，并把组成员信息以及订阅信息发给leader。

第二步：leader开始分配消费方案，指明具体哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案发给coordinator。coordinator接收到分配方案之后会把方案发给各个consumer，这样组内的所有成员就都知道自己应该消费哪些分区了。