zoukankan      html  css  js  c++  java

  • Kafka学习之内核原理剖析

    一、Producer原理分析

    　　1、Producer运行流程

    　　整个生产者客户端由两个线程协调运行，这两个线程分别为主线程和 Sender 线程 (发送线程)。在主线程中由 KafkaProducer 创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器( RecordAccumulator，也称为消息收集器〉中。 Sender 线程负责从RecordAccumulator 中获取消息并将其发送到 Kafka 中 。

    　　　　

    • ProducerRecord: 消息对象。
    • Interceptor: 拦截器，Kafka一共有两种拦截器:生产者拦截器和消费者拦截器。生产者拦截器既可以用来在消息发送前做一些准备工作，比如按照某个规则过滤不符合要求的消息、修改消息的内容等，也可以用来在发送回调逻辑前做一些定制化的需求，比如统计类工作。
    • Serializer: 序列化器，把对象转换成字节数组才能通过网络发送给Kafka。
    • Partitioner: 分区器，消息经过序列化之后就需要确定它发往的分区，如果消息ProducerRecord中指定了partition字段，那么就不需要分区器的作用，因为partition代表的就是所要发往的分区号。如果消息ProducerRecord中没有指定partition字段，那么就需要依赖分区器，根据key这个字段来计算partition的值。分区器的作用就是为消息分配分区。默认分区器:DefaultPartitioner，在partition()方法中定义了主要的分区分配逻辑。如果key不为null，那么默认的分区器会对key进行哈希(采用MurmurHash2算法，具备高运算性能及低碰撞率)，最终根据得到的哈希值取模(所有分区)来计算分区号，拥有相同key的消息会被写入同一个分区。如果key为null，那么消息将会以轮询的方式发往主题内的各个可用分区。
    • RecordAccumulator: 主要用来缓存消息以便Sender线程可以批量发送，进而减少网络传输的资源消耗以提升性能。RecordAccumulator缓存的大小可以通过生产者客户端参数buffer.memory配置，默认值为33554432B，即32M。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候KafkaProducer的send()方法调用要么被阻塞，要么抛出异常，这个取决于参数max.block.ms的配置，此参数的默认值为60000,即60秒。主线程中发送过来的消息都会被迫加到RecordAccumulator的某个双端队列(Deque)中，在RecordAccumulator的内部为每个分区都维护了一个双端队列，队列中的内容就是ProducerBatch，即Deque<ProducerBatch>。消息写入缓存时，追加到双端队列的尾部:Sender读取消息时，从双端队列的头部读取。
    • InFlightRequests: 请求在从Sender线程发往Kafka之前还会保存到InFlightRequests中，InFlightRequests保存对象的具体形式为Map<Nodeld,Deque>，它的主要作用是缓存了已经发出去但还没有收到响应的请求(Nodeld表示broker节点的id编号)。InFlightRequests还提供了许多管理类的方法，并且通过配置参数还可以限制每个连接(也就是客户端与Node之间的连接)最多缓存的请求数。这个配置参数为max.in.flight.requests.per.connection，默认值为5，即每个连接最多只能缓存5个未响应的请求，超过该数值之后就不能再向这个连接发送更多的请求了，除非有缓存的请求收到了响应。InFlightRequests还可以获得leastLoadedNode，即所有Node中负载最小的那一个。选择leastLoadedNode发送请求可以使它能够尽快发出，避免因网络拥塞等异常而影响整体的进度。比如元数据请求，当需要更新元数据时，会先挑选出leastLoadedNode,然后向这个Node发送MetadataRequest请求来获取具体的元数据信息。

    　　　　　　　　　　

    　　2、ACK机制

    　　为保证生产者发送的数据，能可靠的发送到指定的topic，topic的每个partition收到生产者发送的数据后，都需要向生产者发送ack（确认收到），如果生产者收到ack，就会进行下一轮的发送，否则重新发送数据。ACK参数可设置的值为0、1、all。

    • 0：代表producer往集群发送数据不需要等到集群的返回，不确保消息发送成功。安全性最低但是效率最高。
    • 1：代表producer往集群发送数据只要leader应答就可以发送下一条，只确保leader发送成功。
    • all：代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保leader发送成功和所有的副本都完成备份。安全性最高，但是效率最低。

    　　当ACK=all时，Leader和follower（ISR）落盘才会返回ack，会有数据重复现象，如果在leader已经写完成，且follower同步完成，但是在返回ack的出现故障，则会出现数据重复现象；极限情况下，这个也会有数据丢失的情况，比如follower和leader通信都很慢，所以ISR中只有一个leader节点，这个时候，leader完成落盘，就会返回ack，如果此时leader故障后，就会导致丢失数据。

    二、Consumer原理分析

    　　消费者( Consumer )负责订阅 Kafka 中的主题( Topic )，并且从订阅的主题上拉取消息。 在 Kafka 的消费理念中还有一层消费组( Consumer Group)的概念，每个消费者都有一个对应的消费组。当消息发布到主题后，只会被投递给订阅它的每个消费组中的一个消费者 。 每个消费者只能消费所分配到的分区中的消息。换言之，每一个分区只能被一个消费组中的一个消费者所消费 。

    　　　　

    　　1、分区分配策略

    　　Kafka提供了消费者客户端参数partition.assignment.strategy来设置消费者与订阅主题之间的分区分配策略。默认情况下，采用 Range Assignor 分配策略。 Kafka 还提供了另外两种分配策略: RoundRobinAssignor 和 StickyAssignor 。

    • RangeAssignor: RangeAssignor 分配策略的原理是按照消费者总数和分区总数进行整除运算来获得一个跨度，然后将分区按照跨度进行平均分配，以保证分区尽可能均匀地分配给所有的消费者 。
    • RoundRobinAssignor: RoundRobinAssignor 分配策略的原理是将消费组内所有消费者及消费者订阅的所有主题的分区按照字典序排序，然后通过轮询方式逐个将分区依次分配给每个消费者。
    • StickyAssignor: Kafka从 0.11.x 版本开始引入这种分配策略，它主要有两个目的 : 分区的分配要尽可能均匀；分区的分配尽可能与上次分配的保持相同。

    　　2、消息消费

    　　Kafka中的消费是基于拉模式的，拉模式是消费者主动向服务端发起请求来拉取消息。 Kafka 中的消息消费是一个不断轮询的过程，消费者所要做的就是重复地调用 poll()方法 ，而 poll()方法返回的是所订阅的主题(分区)上的一组消息。

    /**
* @see KafkaConsumer#poll(long)
*/
public ConsumerRecords<K, V> poll(long timeout);

    　　timeout 的设置取决于应用程序对响应速度的要求，比如需要在多长时间内将控制权移交给执行轮询的应用线程。可以直接将 timeout 设置为 0 ， 这样 poll()方法会立刻返回，而不管是否己经拉取到了消息。如果应用线程唯一的工作就是从 Kafka 中拉取并消费消息，则可以将这个参数设置为最大值 Long.MAX_VALUE 。

    　　3、位移提交

    　　对于 Kafka 中的分区而言，它的每条消息都有唯一的 offset，用来表示消息在分区中对应的位置 。 对于消费者而言 ，它也有一个 offset 的概念，消费者使用 offset 来表示消费到分区中某个消息所在的位 置。

    　　在每次调用 poll()方法时，它返回的是还没有被消费过的消息集， 要做到这一点，就需要记录上一次消费时的消费位移 。 消费位移存储在 Kafka 内 部的主题 consumer offsets 中 。 这里把将消费位移存储起来(持久化)的动作称为提交 ，消费者在消费完消息之后需要执行消费位移的提交。

     　　　　

    　　在Kafka中默认的消费位移的提交方式是自动提交，这个由消费者客户端参数enable.auto.commit配置，默认值为true。当然这个默认的自动提交不是每消费一条消息就提交一次，而是定期提交，这个定期的周期时间由户端参数auto.commit.interval.ms配置，默认值为5秒，此参数生效的前提是enable.auto.commit参数为true。

    　　每个consumer会定期将自己消费分区的offset提交给kafka内部topic: consumer_offsets，提交过去的时候，key是consumerGroupId+topic+分区号，value就是当前offset的值，kafka会定期清理topic里的消息，最后就保留最新的那条数据。因为consumer_offsets可能会接收高并发的请求，kafka默认给其分配50个分区(可以通过offsets.topic.num.partitions设置)，这样可以通过加机器的方式扩大并发。

    　　4、Rebalance机制

    　　rebalance是指分区的所属权从一个消费者转移到另一消费者的行为，它为消费组具备高可用性和伸缩性提供保障。rebalance发生期间，消费组内的消费者是无法读取消息的。当一个分区被重新分配给另一个消费者时，消费者当前的状态也会丢失。

    　　如下情况可能会触发消费者rebalance

    • consumer所在服务器重启或宕机了；
    • 动态给topic增加了分区；
    • 消费组订阅了更多的topic。
    public interface ConsumerRebalanceListener{
　　// 在rebalance开始之前和消费者停止读取消息之后被调用。
    　　// 可以通过这个回调方法    来处理消费位移的提交，以此来避免一些不必要的重复消费现象的发生。
　　// 参数 partitions表示rebalance前所分配到的分区。
　　void onPartitionsRevoked(Collection<TopicPartition> partitions); 　　
    　　//     在重新分配分区之后和消费者开始读取消费之前被调用，参数partitions表示再均衡后所分配到的分区。
　　void onPartitionsAssigned(Collection<TopicPartition> partitions);
}

    　　5、消费者协调器和组协调器

    　　每个消费组的子集在服务端对应一个GroupCoordinator对其进行管理，GroupCoordinator是Kafka服务端中用于管理消费组的组件。而消费者客户端中的ConsumerCoordinator组件负责与GroupCoordinator进行交互。ConsumerCoordinator与GroupCoordinator之间最重要的职责就是负责执行消费者再均衡的操作，包括前面提及的分区分配的工作也是在rebalance期间完成的。一共有如下几种情形会触发再均衡的操作:

    • 有新的消费者加入消费组。
    • 有消费者宕机下线。消费者并不一定需要真正下线，例如遇到长时间的GC、网络延迟导致消费者长时间未向GroupCoordinator发送心跳等情况时，GroupCoordinator会认为消费者己经下线。
    • 有消费者主动退出消费组(发送LeaveGroupRequest请求)。比如客户端调用了unsubscrible()方法取消对某些主题的订阅。
    • 消费组所对应的GroupCoorinator节点发生了变更。
    • 消费组内所订阅的任一主题或者主题的分区数量发生变化。

    　　当有消费者加入消费组时，消费者、消费组及组协调器之间会经历一下几个阶段。整个rebalance过程如下:

    　　第一阶段:选择组协调器(FINDCOORDINATOR)

    　　组协调器GroupCoordinator: 每个consumergroup都会选择一个broker作为自己的组协调器coordinator，负责监控这个消费组里的所有消费者的心跳，以及判断是否宕机，然后开启消费者rebalance。consumergroup中的每个consumer启动时会向kafka集群中的某个节点(负载最小的节点：InFlightRequests有记录)发送FindCoordinatorRequest请求来查找对应的组协调器GroupCoordinator，并跟其建立网络连接。

    　　组协调器选择方式: 通过如下公式可以选出consumer消费的offset要提交到consumer_offsets的哪个分区，这个分区leader对应的broker就是这个consumergroup的coordinator公式: hash(consumergroupid)%consumer_offsets 主题的分区数

    　　第二阶段:加入消费组(JOINGROUP)

    　　在成功找到消费组所对应的GroupCoordinator之后就进入加入消费组的阶段，在此阶段的消费者会向GroupCoordinator发送JoinGroupRequest请求，并处理响应。然后GroupCoordinator从一个consumergroup中选择第一个加入group的consumer作为leader(消费组协调器)，把consumergroup情况发送给这个leader，接着这个leader会负责制定分区方案。

    　　选举分区分配策略，根据各个消费者呈报的分配策略选举，过程如下:

    • 收集各个消费者支持的所有分配策略，组成候选集candidates。
    • 每个消费者从候选集candidates中找出第一个自身支持的策略，为这个策略投上一票。
    • 计算候选集中各个策略的选票数，选票数最多的策略即为当前消费组的分配策略。

    　　每个消费者都向GroupCoordinator发送JoinGroupRequest请求，其中携带了各自提案的分配策略和订阅信息。JoinGroupResonse回执中包含GroupCoordinator中投票选举出的分配策略的信息,并且只有leader消费者的回执中包含每个消费者的订阅信息。

    　　第三阶段:同步消费组(SYNCGROUP)

    　　consumerleader通过给GroupCoordinator发送SyncGroupRequest，接着GroupCoordinator就把分区方案下发给各个consumer，他们会根据指定分区的leaderbroker进行网络连接以及消息消费。

    　　6、多线程

    　　KafkaConsumer是非线程安全的，KafkaConsumer中定义了一个acquire()方法，用来检测当前是否只有一个线程在操作，若有其他线程正在操作则会抛出ConcurrentModifcationException异常。

    三、Broker原理分析

    　　1、分区管理

    　　优先副本的选举：Kafka集群的一个broker中最多只能有它的一个副本。分区使用多副本机制来提升可靠性，但只有leader副本对外提供读写服务，而follower副本只负责在内部进行消息的同步。如果一个分区的leader副本不可用，那么就意味着整个分区变得不可用，此时就需要Kafka从剩余的follower副本中挑选一个新的leader副本来继续对外提供服务。为了能够有效地治理负载失衡的情况，Kafka引入了优先副本(preferredreplica)的概念。所谓的优先副本是指在AR集合列表中的第一个副本。优先副本的选举是指通过一定的方式促使优先副本选举为leader副本，以此来促进集群的负载均衡，这一行为也可以称为“分区平衡”。

    　　　　　　

    bin/kafka-topics.sh --describe --zookeeper localhost:2181 m --topic my-replicated-topic2　　//查看分区信息

    　　结果如下所示(包含了部分Broker下线以及恢复的过程)：

    HoudeMacBook-Pro:kafka_2.12-2.6.0 houjing$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic　//查看分区信息
Topic: my-replicated-topic    PartitionCount: 3    ReplicationFactor: 3    Configs: 
    Topic: my-replicated-topic    Partition: 0    Leader: 2    Replicas: 2,0,1    Isr: 2,0,1
    Topic: my-replicated-topic    Partition: 1    Leader: 0    Replicas: 0,1,2    Isr: 0,1,2
    Topic: my-replicated-topic    Partition: 2    Leader: 1    Replicas: 1,2,0    Isr: 1,2,0
HoudeMacBook-Pro:kafka_2.12-2.6.0 houjing$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic //kill Broker1后分区信息
Topic: my-replicated-topic    PartitionCount: 3    ReplicationFactor: 3    Configs: 
    Topic: my-replicated-topic    Partition: 0    Leader: 2    Replicas: 2,0,1    Isr: 2,0
    Topic: my-replicated-topic    Partition: 1    Leader: 0    Replicas: 0,1,2    Isr: 0,2
    Topic: my-replicated-topic    Partition: 2    Leader: 2    Replicas: 1,2,0    Isr: 2,0
HoudeMacBook-Pro:kafka_2.12-2.6.0 houjing$ bin/kafka-server-start.sh -daemon config/server2.properties   //重启broker1
HoudeMacBook-Pro:kafka_2.12-2.6.0 houjing$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic //查看分区信息
Topic: my-replicated-topic    PartitionCount: 3    ReplicationFactor: 3    Configs: 
    Topic: my-replicated-topic    Partition: 0    Leader: 2    Replicas: 2,0,1    Isr: 2,0,1
    Topic: my-replicated-topic    Partition: 1    Leader: 0    Replicas: 0,1,2    Isr: 0,2,1
    Topic: my-replicated-topic    Partition: 2    Leader: 2    Replicas: 1,2,0    Isr: 2,0,1

    　　在Kafka中可以提供分区自动平衡的功能，与此对应的broker端参数是auto.leader.rebalance.enable，此参数的默认值为true，即默认情况下此功能是开启的。如果开启分区自动平衡的功能，则Kafka的控制器会启动一个定时任务，这个定时任务会轮询所有的broker节点，计算每个broker节点的分区不平衡率 (broker中的不平衡率=非优先副本的leader个数/分区总数)是否超过leader.imbalance.per.broker.percentage参数配置的比值，默认值为10%，如果超过设定的比值则会自动执行优先副本的选举动作以求分区平衡。执行周期由参数leader.imbalance.check.interval.seconds控制，默认值为300秒。Kafka中kafka-perferred-replica-election.sh脚本提供了对分区leader副本进行重新平衡的功能。

    • 分区重分配：Kafka提供了kafka-reassign-partitions.sh脚本来执行分区重分配的工作，它可以在集群扩容、broker节点失效的场景下对分区进行迁移。
    • 失效副本：正常情况下，分区的所有副本都处于ISR集合，但是难免会有异常发生的情况，从而某些副本被剥离出ISR集合。当follower副本将leader副本LEO(Log End Offset)之前的日志全部同步时，则认为该fillower副本已经追赶上leader副本，此时更新该副本的lastAoughtUpTimeMs标识，Kafka的副本管理器会启动一个副本过期检测的定时任务，而这个定时任务会定时检查当前时间与副本的lastAoughtUpTimeMs差值是否大于参数replica.time.max.ms指定的值，默认10000。

    　　2、控制器Controller

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

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

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

    # brokerid表示称为控制器的broker的id编号，timestamp表示竞选为控制器时的时间戳 
    {"version":1,"brokerid":0,"timestamp":"1574831950372"}

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

    • Partition副本leader的选举

    　　分区leader副本的选举由控制器负责具体实施。当创建分区(创建主题或增加分区都有创建分区的动作)或分区上线(比如分区中原先的leader副本下线，此时分区需要选举一个新的leader上线来对外提供服务)的时候都需要执行leader的选举动作。这种策略的基本思路是按照AR集合中副本的顺序查找第一个存活的副本，并且这个副本在JSR集合中。一个分区的AR集合在分配的时候就被指定，并且只要不发生重分配的情况，集合内部副本的顺序是保持不变的，而分区的ISR集合中副本的顺序可能会改变。

    ##在zookeeper中查看各broker节点信息
[zk: localhost:2181(CONNECTED) 7] get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://10.1.140.146:9092"]
    　　,"jmx_port":-1,"host":"10.1.140.146","timestamp":"1603270074840","port":9092,"version":4}
[zk: localhost:2181(CONNECTED) 8] get /brokers/ids/1　　##broker1 被kill后
org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids/1
[zk: localhost:2181(CONNECTED) 9] get /brokers/ids/2
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://10.1.140.146:9094"]
    　　,"jmx_port":-1,"host":"10.1.140.146","timestamp":"1603270074178","port":9094,"version":4}
    ## 在zookeeper中查看分区信息
[zk: localhost:2181(CONNECTED) 15] get /brokers/topics/my-replicated-topic/partitions/2/state
{"controller_epoch":6,"leader":2,"version":1,"leader_epoch":1,"isr":[2,0,1]}
[zk: localhost:2181(CONNECTED) 16] get /brokers/topics/my-replicated-topic/partitions/1/state
{"controller_epoch":6,"leader":0,"version":1,"leader_epoch":1,"isr":[0,2,1]}
[zk: localhost:2181(CONNECTED) 17] get /brokers/topics/my-replicated-topic/partitions/0/state
{"controller_epoch":6,"leader":2,"version":1,"leader_epoch":1,"isr":[2,0,1]}

    　　3、LEO与HW高水位

    　　整个消息追加的过程可以概括如下:

    • 生产者客户端发送消息至leader副本中。
    • 消息被迫加到leader副本的本地日志，并且会更新日志的偏移量。
    • follower副本向leader副本请求同步数据。
    • leader副本所在的服务器读取本地日志，并更新对应拉取的follower副本的信息。
    • leader副本所在的服务器将拉取结果返回给follower副本。
    • follower副本收到leader副本返回的拉取结果，将消息追加到本地日志中，并更新日志的偏移量信息。

    　　在Kafka中，高水位(HighWatermark)的作用主要有2个:

    • 定义消息可见性，即用来标识分区下的哪些消息是可以被消费者消费的。
    • 帮助Kafka完成副本同步。

    　　　　　　

    　　在分区高水位以下的消息被认为是已提交消息，反之就是未提交消息。消费者只能消费已提交消息，即图中位移小于15的所有消息(不考虑事务)。位移值等于高水位的消息也属于未提交消息。

    　　日志末端位移(LEO：Log End Offset)表示副本写入下一条消息的位移值。同一个副本对象，其高水位值不会大于LEO值。高水位和LEO是副本对象的两个重要属性。Kafka所有副本都有对应的高水位和LEO值，而不仅仅是Leader本。只不过Leader副本比较特殊，Kafka使用Leader副本的高水位来定义所在分区的高水位。换句话说，分区的高水位就是其Leader副本的高水位。

    　　高水位更新机制

    　　在Leader副本所在的Broker上，还保存了其他Follower副本的LEO值。

    　　　　 　

    　　Broker0上保存了某分区的Leader副本和所有Follower副本的LEO值，而Broker1上仅仅保存了该分区的某个Follower副本。Kafka把Broker0上保存的这些Follower副本又称为远程副本(RemoteReplica)。Kafka副本机制在运行过程中，会更新Broker1上Follower副本的高水位和LEO值，同时也会更新Broker0上Leader副本的高水位和LEO以及所有远程副本的LEO，但它不会更新远程副本的高水位值，也就是图中标记为灰色的部分。

    　　Broker0上保存这些远程副本主要作用是: 帮助Leader副本确定其高水位，也就是分区高水位。

    　　与Leader副本保持同步的判断的条件有两个:

    • 该远程Follower副本在ISR中。
    • 该远程Follower副本LEO值落后于Leader副本LEO值的时间，不超过Broker端参数replica.lag.time.max.ms的值。如果使用默认值的话，就是不超过10秒。

    　　取一个partition对应的ISR中最小的LEO(log-end-offset)作为HW，consumer最多只能消费到HW所在的位置。对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更HW，此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。对于来自内部broker的读取请求，没有HW的限制。

    　　副本同步机制

    　　当producer生产消息至broker后，ISR以及HW和LEO的流转过程:

    　　　　　　

    　　Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。事实上，同步复制要求所有能工作的follower都复制完，这条消息才会被commit，这种复制方式极大的影响了吞吐率。而异步复制方式下，follower异步的从leader复制数据，数据只要被leader写入log就被认为已经commit，这种情况下如果follower都还没有复制完，落后于leader时，突然leader宕机，则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。

    　　结合HW和LEO看下acks=1的情况:

    　　 　　　　

    四、日志存储

    　　Kafka一个分区的消息数据对应存储在一个文件夹下，以topic名称+分区号命名，kafka规定了一个分区内的.log文件最大为1G。

    　　　　　　

    # 部分消息的offset索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的offset到 index文件，
# 如果要定位消息的offset会先在这个文件里快速定位，再去log文件里找具体消息 
00000000000000000000.index
# 消息存储文件，主要存offset和消息体
# LogSegment 的基准偏移量为 0，对应的日志文件为 00000000000000000000.log 
    00000000000000000000.log
# 消息的发送时间索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的发送时间戳与对 应的offset到timeindex文件，
# 如果需要按照时间来定位消息的offset，会先在这个文件里查找 00000000000000000000.timeindex
000000000000009900000.index
000000000000009900000.log
000000000000009900000.timeindex

    　　KafkaBroker有一个参数，log.segment.bytes，限定了每个日志段文件的大小，最大就是1GB。一个日志段文件满了，就自动开一个新的日志段文件来写入，避免单个文件过大，影响文件的读写性能，这个过程叫做logrolling，正在被写入的那个日志段文件，叫做activelogsegment。

    　　1、日志索引

    　　每个日志分段文件对应了两个索引文件，主要用来提高查找消息的效率。偏移量索引文件用来建立消息偏移量offset到物理地址之间的映射关系，方便快速定位消息所在的物理文件位置;时间戳索引文件则根据指定的时间戳timestamp来查找对应的偏移量信息。Kafka中的索引文件以稀疏索引(sparseindex)的方式构造消息的索引，它并不保证每个消息在索引文件中都有对应的索引页。每当写入一定量(由broker端参数log.index.interval.bytes指定，默认值为4096，即4KB的消息时，偏移量索引文件和时间戳索引文件分别增加一个偏移量索引项和时间戳索引项，增大或减小log.index.interval.bytes的值，对应地可以增加或缩小索引项的密度。

    　　稀疏索引通过MappedByteBuffer将索引文件映射到内存中，以加快索引的查询速度。偏移量索引文件中的偏移量是单调递增的，查询指定偏移量时，使用二分查找法来快速定位偏移量的位置，如果指定的偏移量不在索引文件中，则会返回小于指定偏移量的最大偏移量。时间戳索引文件中的时间戳也保持严格的单调递增，查询指定时间戳时，也根据二分查找法来查找不大于该时间戳的最大偏移量.

    • 偏移量索引:每个索引项占用8个字节，分为两个部分。
      • relativeOffset: 相对偏移量，表示消息相对于baseOffset的偏移量，占用4个字节，当前索引文件的文件名即为baseOffset的值。
      • position: 物理地址，也就是消息在日志分段文件中对应的物理位置，占用4个字节。
    • 时间戳索引:每个索引项占用12个字节，分为两个部分。
      • timestamp:当前日志分段最大的时间戳，占用8个字节。
      • relativeOffset:时间戳所对应的消息的相对偏移量，占用4个字节。

    　　如果broker端参数log.message.timestamp.type设置为LogAppendTime，那么消息的时间戳必定能够保持单调递增;相反，如果是CreateTime类型则无法保证。

    　　2、零拷贝

    　　Kafka使用零拷贝技术来提升性能。所谓的零拷贝是指将数据直接从磁盘文件复制到网卡设备中，而不需要经由应用程序之手。零拷贝大大提高了应用程序的性能，减少了内核和用户模式之间的上下文切换。对Linux操作系统而言，零拷贝技术依赖于底层的sendfile()方法实现。对应于Java语言，FileChannal.transferTo()方法的底层实现就是sendfile()方法。

    　　　　　　
  • 相关阅读:
    手机维修费杂乱 消协命令设立扶植行业标准
    ImageList控件的使用方法
    三峡实习记之一
    [原创]2天的搜索经验对IIS的问题的全面解决方案
    亲历虚拟机安装Ubuntu Linux系统
    新的一年,新的开始
    VS2003中实现身份验证的探索经验
    [原创]VS2003中ASP.NET实现自定义用户导航控件
    全面控制Windows任务栏
    单刀赴广州
  • 原文地址：https://www.cnblogs.com/jing99/p/13861149.html
Copyright © 2011-2022 走看看