zoukankan      html  css  js  c++  java
  • Kafka笔记

    第1章 Kafka简介
    1.1 kafka起源
    Kafka是由LinkedIn开发并开源的分布式消息系统,2012年捐赠给Apache基金会,采用Scala语言,运行在JVM中,最新版本1.0.0。
     
    1.2 kafka设计目标
    Kafka是一种分布式的,基于发布/订阅的消息系统。主要设计目标如下:
    ①以时间复杂度O(1)的方式提供消息持久化能力,即使对TB级以上数据也能保证常数时间复杂度的访问性能。
    ②高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒10万条以上消息的传输。
    ③支持Kafka Server间的消息分区,及分布式消费,同时保证每个Partition内的消息顺序传输。
    ④同时支持离线数据处理和实时数据处理。
    ⑤Scale out:支持在线水平扩展。
     
    1.3 为何使用消息系统
    消息系统有如下好处:解耦、冗余、扩展性、灵活性&峰值处理能力、可恢复性、顺序保证、缓冲、异步通信。
     
    1.4 RabbitMQ、ActiveMQ和Kafka的区别
     
     
    第2章 Kafka架构
    2.1 kafka术语
    ①Topic
    用于划分Message的逻辑概念,一个Topic可以分布在多个Broker上。
    ②Partition
    是Kafka中横向扩展和一切并行化的基础,是物理上的概念,每个Topic都至少被切分为1个Partition。
    ③Offset
    消息在Partition中的编号,编号顺序不跨Partition。
    ④Consumer
    用于从Broker中取出/消费Message。
    ⑤Consumer Group
    每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name,若不指定group name则属于默认的group)。
    ⑥Producer
    用于往Broker中发送/生产Message。
    ⑦Replication
    Kafka支持以Partition为单位对Message进行冗余备份,每个Partition都可以配置至少1个Replication(当仅1个Replication时即仅该Partition本身)。
    ⑧Leader
    每个Replication集合中的Partition都会选出一个唯一的Leader,所有的读写请求都由Leader处理。其他Replicas从Leader处把数据更新同步到本地,过程类似大家熟悉的MySQL中的Binlog同步。
    ⑨Broker
    Kafka集群包含一个或多个服务器,这种服务器被称为broker。Kafka中使用Broker来接受Producer和Consumer的请求,并把Message持久化到本地磁盘。每个Cluster当中会选举出一个Broker来担任Controller,负责处理Partition的Leader选举,协调Partition迁移等工作。
    ⑩ISR(In-Sync Replica)
    是Replicas的一个子集,表示目前Alive且与Leader能够“Catch-up”的Replicas集合。由于读写都是首先落到Leader上,所以一般来说通过同步机制从Leader上拉取数据的Replica都会和Leader有一些延迟(包括了延迟时间和延迟条数两个维度),任意一个超过阈值都会把该Replica踢出ISR。每个Partition都有它自己独立的ISR。
     
    2.2 kafka拓扑结构
    一个典型的Kafka集群中包含多个Producer,多个broker(Kafka支持水平扩展,一般broker数量越多,集群吞吐率越高),多个Consumer Group,以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置,选举leader,以及在Consumer Group发生变化时进行rebalance(再平衡)。Producer使用push模式将消息发布到broker,Consumer使用pull模式从broker订阅并消费消息。
    2.3 kafka高吞吐量的黑科技
    kafka实现高性能吞吐的秘密主要有三点:
    ①磁盘顺序IO
    由于消息系统读写的特殊性,Kafka在磁盘上只做Sequence I/O。关于磁盘I/O的性能,Kafka官方给出的测试数据(Raid-5,7200rpm)为:Sequence I/O: 600MB/s;Random I/O: 100KB/s 。
    ②PageCache
    操作系统提供的PageCache功能。当上层有写操作时,操作系统只是将数据写入PageCache,同时标记Page属性为Dirty。当读操作发生时,先从PageCache中查找,如果发生缺页才进行磁盘调度,最终返回需要的数据。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。同时如果有其他进程申请内存,回收PageCache的代价又很小,所以现代的OS都支持PageCache。
    直接使用PageCache功能而不是在JVM内部缓存数据,可带来如下好处:避免频繁GC(当Full GC时会影响可用性)、内存占用空间翻倍(Java堆和PageCache都有同一份数据)、避免Kafka进程重启Java堆里数据失效(PageCache仍然可用)。
    ③Sendfile技术
    数据从磁盘读出并通过网络Socket发送出去这个过程,如果采用Sendfile技术,可以避免数据在内核空间和用户空间的无用拷贝。
    如果没有sendfile技术,数据流如下:
    1. OS 从硬盘把数据读到内核区的PageCache。
    2. 用户进程把数据从内核区Copy到用户区。
    3. 然后用户进程再把数据写入到Socket,数据流入内核区的Socket Buffer上。
    4. OS 再把数据从Buffer中Copy到网卡的Buffer上,这样完成一次发送。
    经过Sendfile优化后,整个I/O过程如下:
    ④offset无锁消费
    Kafka会为每一个Consumer Group保留一些metadata信息——当前消费的消息的position,也即offset。这个offset由Consumer控制。正常情况下Consumer会在消费完一条消息后递增该offset。当然,Consumer也可将offset设成一个较小的值,重新消费一些消息。因为offet由Consumer控制,所以Kafka broker是无状态的,它不需要标记哪些消息被哪些消费过,也不需要通过broker去保证同一个Consumer Group只有一个Consumer能消费某一条消息,因此也就不需要锁机制,这也为Kafka的高吞吐率提供了有力保障。  
     
     
    第3章 Topic & Partition
    3.1 Topic和Partition简介
    Topic在逻辑上可以被认为是一个queue,每条消息都必须指定它的Topic,可以简单理解为必须指明把这条消息放进哪个queue里。为了使得Kafka的吞吐率可以线性提高,物理上把Topic分成一个或多个Partition,每个Partition在物理上对应一个文件夹,该文件夹下存储这个Partition的所有消息和索引文件。
    每个日志文件都是一个log entry序列,每个log entry包含一个4字节整型数值(值为N+5),1个字节的”magic value”,4个字节的CRC校验码,其后跟N个字节的消息体。每条消息都有一个当前Partition下唯一的64字节的offset,它指明了这条消息的起始位置。
    这个log entry并非由一个文件构成,而是分成多个segment,每个segment以该segment第一条消息的offset命名并以“.kafka”为后缀。另外会有一个索引文件,它标明了每个segment下包含的log entry的offset范围,如下图所示:
    因为每条消息都被append到该Partition中,属于顺序写磁盘,因此效率非常高。
     
    3.2 消息删除
    对于传统的message queue而言,一般会删除已经被消费的消息,而Kafka集群会保留所有的消息,无论其被消费与否。当然,因为磁盘限制,不可能永久保留所有数据(实际上也没必要),因此Kafka提供两种策略删除旧数据。一是基于时间,二是基于Partition文件大小。
    因为Kafka读取特定消息的时间复杂度为O(1),即与文件大小无关,所以这里删除过期文件与提高Kafka性能无关。
     
    3.3 Partition特点
    Partition是Kafka可以很好的横向扩展和提供高并发处理以及实现Replication的基础。
    ①扩展性方面
    首先,Kafka允许Partition在集群内的Broker之间任意移动,以此来均衡可能存在的数据倾斜问题。其次,Partition支持自定义的分区算法,例如可以将同一个Key的所有消息都路由到同一个Partition上去。 同时Leader也可以在In-Sync的Replica中迁移。由于针对某一个Partition的所有读写请求都是只由Leader来处理,所以Kafka会尽量把Leader均匀的分散到集群的各个节点上,以免造成网络流量过于集中。
    并发方面
    任意Partition在某一个时刻只能被一个Consumer Group内的一个Consumer消费(反过来一个Consumer则可以同时消费多个Partition),Kafka非常简洁的Offset机制最小化了Broker和Consumer之间的交互,这使Kafka并不会像同类其他消息队列一样,随着下游Consumer数目的增加而成比例的降低性能。此外,如果多个Consumer恰巧都是消费时间序上很相近的数据,可以达到很高的PageCache命中率,因而Kafka可以非常高效的支持高并发读操作,实践中基本可以达到单机网卡上限。
     
    3.4 Partition过多引起的问题
    Partition的数量并不是越多越好,Partition的数量越多,平均到每一个Broker上的数量也就越多,会导致如下问题:
    ①Broker宕机(Network Failure, Full GC)的情况下,需要由Controller来为所有宕机的Broker上的所有Partition重新选举Leader,Partition数量太多会Leader选举耗时过长。
    ②Controller角色的Broker宕机,首先要进行的是重新任命一个Broker作为Controller。新任命的Controller要从Zookeeper上获取所有Partition的Meta信息,Partition数量太多会导致获取时间过长。另外获取完后还要进行上一步的Leader选举。
    ③在Broker端,对Producer和Consumer都使用了Buffer机制。其中Buffer的大小是统一配置的,数量则与Partition个数相同。如果Partition个数过多,会导致Producer和Consumer的Buffer内存占用过大。
     
    Partition规划提示:
    ①Partition的数量尽量提前预分配,虽然可以在后期动态增加Partition,但是会冒着可能破坏Message Key和Partition之间对应关系的风险。
    ②Replica的数量不要过多,如果条件允许尽量把Replica集合内的Partition分别调整到不同的Rack。
    ③尽一切努力保证每次停Broker时都可以Clean Shutdown,否则问题就不仅仅是恢复服务所需时间长,还可能出现数据损坏或其他很诡异的问题。
     
     
    第4章 Producer
    4.1 Partition分区优点
    消息生产时,采用Partition分区的优点如下:
    ①实现消息生产时负载均衡
    Producer发送消息到broker时,会根据Paritition机制选择将其存储到哪一个Partition。如果Partition机制设置合理,所有消息可以均匀分布到不同的Partition里,这样就实现了负载均衡。
    ②消息并行写入提高吞吐量
    如果一个Topic对应一个文件,那这个文件所在的机器I/O将会成为这个Topic的性能瓶颈,而有了Partition后,不同的消息可以并行写入不同broker的不同Partition里,极大的提高了吞吐量。
     
    4.2 Producer消息路由
    在发送一条消息时,可以指定这条消息的key,Producer根据这个key和Partition机制来判断应该将这条消息发送到哪个Parition。Paritition机制可以通过指定Producer的paritition. class这一参数来指定,该class必须实现kafka.producer.Partitioner接口。
     
    4.2 Producer端优化
    Producer端的优化思路有两种:批量发布、同步变异步,关闭ACK全速发送。
    ①批量发布
    Kafka系统默认支持MessageSet,把多条Message自动地打成一个Group后发送出去,均摊后拉低了每次通信的RTT(Round-Trip Time 往返时延)。而且在组织MessageSet的同时,还可以把数据重新排序,从爆发流式的随机写入优化成较为平稳的线性写入。
    Producer支持End-to-End的压缩。数据在本地压缩后放到网络上传输,在Broker一般不解压(除非指定要Deep-Iteration),直至消息被Consume之后在客户端解压。
     
     
    第5章 Consumer
    5.1 Consumer Group
    使用Consumer high level API时,同一Topic的一条消息只能被同一个Consumer Group内的一个Consumer消费,但多个Consumer Group可同时消费这一消息。
    这是Kafka用来实现一个Topic消息的广播(发给所有的Consumer)和单播(发给某一个Consumer)的手段。一个Topic可以对应多个Consumer Group。如果需要实现广播,只要每个Consumer有一个独立的Group就可以了。要实现单播只要所有的Consumer在同一个Group里。用Consumer Group还可以将Consumer进行自由的分组而不需要多次发送消息到不同的Topic。
    Kafka的设计理念之一就是同时提供离线处理和实时处理。只需要保证离线和实时操作所使用的Consumer属于不同的Consumer Group即可。
     
    5.2 Push vs. Pull
    push模式很难适应消费速率不同的消费者,因为消息发送速率是由broker决定的。push模式的目标是尽可能以最快速度传递消息,但是这样很容易造成Consumer来不及处理消息,典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据Consumer的消费能力以适当的速率消费消息。
    对于Kafka而言,pull模式更合适。pull模式可简化broker的设计,Consumer可自主控制消费消息的速率,同时Consumer可以自己控制消费方式——即可批量消费也可逐条消费,同时还能选择不同的提交方式从而实现不同的传输语义。
     
    5.3 Kafka delivery guarantee
    Kafka的delivery guarantee有三种:
    ①At most once 最多一次。 消息可能会丢,但绝不会重复传输
    读完消息先commit再处理消息。这种模式下,如果Consumer在commit后还没来得及处理消息就crash了,下次重新开始工作后就无法读到刚刚已提交而未处理的消息,这就对应于At most once。
    ②At least one 最少一次。 消息绝不会丢,但可能会重复传输
    读完消息先处理再commit。这种模式下,如果在处理完消息之后commit之前Consumer crash了,下次重新开始工作时还会处理刚刚未commit的消息,实际上该消息已经被处理过了。这就对应于At least once。
    ③Exactly once 精确一次。 每条消息肯定会被传输一次且仅传输一次,很多时候这是用户所想要的。
    如果一定要做到Exactly once,就需要协调offset和实际操作的输出。经典的做法是引入两阶段提交。如果能让offset和操作输入存在同一个地方,会更简洁和通用。
     
    Kafka默认保证At least once,并且允许通过设置Producer异步提交来实现At most once。而Exactly once要求与外部存储系统协作,幸运的是Kafka提供的offset可以非常直接非常容易得使用这种方式。
     
    5.4 High level和Low level
    Consumer API分为High level和Low level两种。前一种重度依赖Zookeeper,所以性能差一些且不自由,但是超省心。第二种不依赖Zookeeper服务,无论从自由度和性能上都有更好的表现,但是所有的异常(Leader迁移、Offset越界、Broker宕机等)和Offset的维护都需要自行处理。
     
    5.5 如何保证kafka的高容错性?
    ①producer不使用批量接口,并采用同步模型持久化消息。
    ②consumer不采用批量化,每消费一次就更新offset。
     
     
    第6章 Kafka高可用
    6.1 为什么需要Replication
    在没有Replication的情况下,一旦某机器宕机或者某个Broker停止工作则其上所有的Partition数据都不可被消费,会造成整个系统的可用性降低。
     
    6.2 Replica之间为什么需要Leader Election
    引入Replication之后,同一个Partition可能会有多个Replica,而这时需要在这些Replica中选出一个Leader,Producer和Consumer只与这个Leader交互,其它Replica作为Follower从Leader中复制数据。
    因为需要保证同一个Partition的多个Replica之间的数据一致性和有序性。
     
    6.3 Replica均匀分配算法
    为了更好的做负载均衡,Kafka尽量将所有的Partition均匀分配到整个集群上。一个典型的部署方式是一个Topic的Partition数量大于Broker的数量。同时为了提高Kafka的容错能力,也需要将同一个Partition的Replica尽量分散到不同的机器。
    Kafka分配Replica的算法如下:
    ①将所有Broker(假设共n个Broker)和待分配的Partition排序。
    ②将第i个Partition分配到第(i mod n)个Broker上。
    ③将第i个Partition的第j个Replica分配到第((i + j) mod n)个Broker上。
     
    6.4 Propagate消息
    Producer在发布消息到某个Partition时,先通过Zookeeper找到该Partition的Leader,然后无论该Topic的Replication Factor为多少(也即该Partition有多少个Replica),Producer只将该消息发送到该Partition的Leader。Leader会将该消息写入其本地Log。每个Follower都从Leader pull数据。这种方式上,Follower存储的数据顺序与Leader保持一致。Follower在收到该消息并写入其Log后,向Leader发送ACK。一旦Leader收到了ISR中的所有Replica的ACK,该消息就被认为已经commit了,Leader将增加HW(HighWaterMark,Partition的高水位)并且向Producer发送ACK。
    为了提高性能,每个Follower在接收到数据后就立马向Leader发送ACK,而非等到数据写入Log中。因此,对于已经commit的消息,Kafka只能保证它被存于多个Replica的内存中,而不能保证它们被持久化到磁盘中,也就不能完全保证异常发生后该条消息一定能被Consumer消费。但考虑到这种场景非常少见,可以认为这种方式在性能和数据持久化上做了一个比较好的平衡。在将来的版本中,Kafka会考虑提供更高的持久性。
    Consumer读消息也是从Leader读取,只有被commit过的消息(offset低于HW的消息)才会暴露给Consumer。
     
    6.4 ACK前需要保证有多少个备份
    Kafka存活包含两个条件,一是它必须维护与Zookeeper的session(这个通过Zookeeper的Heartbeat机制来实现)。二是Follower必须能够及时将Leader的消息复制过来,不能“落后太多”。
    Leader会跟踪与其保持同步的Replica列表,该列表称为ISR(即in-sync Replica)。如果一个Follower宕机,或者落后太多,Leader将把它从ISR中移除。这里所描述的“落后太多”指Follower复制的消息落后于Leader后的条数超过预定值,或者Follower超过一定时间未向Leader发送fetch请求。
    Kafka的复制机制既不是完全的同步复制,也不是单纯的异步复制。事实上,同步复制要求所有能工作的Follower都复制完,这条消息才会被认为commit,这种复制方式极大的影响了吞吐率(高吞吐率是Kafka非常重要的一个特性)。而异步复制方式下,Follower异步的从Leader复制数据,数据只要被Leader写入log就被认为已经commit,这种情况下如果Follower都复制完都落后于Leader,而如果Leader突然宕机,则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。Follower可以批量的从Leader复制数据,这样极大的提高复制性能(批量写磁盘),极大减少了Follower与Leader的差距。
    一条消息只有被ISR里的所有Follower都从Leader复制过去才会被认为已提交。这样就避免了部分数据被写进了Leader,还没来得及被任何Follower复制就宕机了,而造成数据丢失(Consumer无法消费这些数据)。而对于Producer而言,它可以选择是否等待消息commit,这可以通过request.required.acks来设置。这种机制确保了只要ISR有一个或以上的Follower,一条被commit的消息就不会丢失。
     
    6.5 Leader Election算法
    一种非常常用的Leader Election的方式是“Majority Vote”(“少数服从多数”),但Kafka并未采用这种方式。这种模式下,如果我们有2f+1个Replica(包含Leader和Follower),那在commit之前必须保证有f+1个Replica复制完消息,为了保证正确选出新的Leader,fail的Replica不能超过f个。因为在剩下的任意f+1个Replica里,至少有一个Replica包含有最新的所有消息。 
    Majority Vote优点:系统的延迟时间只取决于最快的几个Broker,而非最慢那个。
    Majority Vote缺点:为了保证较高的容错程度,必须要有大量的Replica,而大量的Replica又会在大数据量下导致性能的急剧下降。
    为了保证Leader Election的正常进行,它所能容忍的fail的follower个数比较少。如果要容忍1个follower挂掉,必须要有3个以上的Replica,如果要容忍2个Follower挂掉,必须要有5个以上的Replica。这就是这种算法更多用在Zookeeper这种共享集群配置的系统中,而很少在需要存储大量数据的系统中使用的原因。
     
    Kafka在Zookeeper中动态维护了一个ISR(in-sync replicas),这个ISR里的所有Replica都跟上了leader,只有ISR里的成员才有被选为Leader的可能。在这种模式下,对于f+1个Replica,一个Partition能在保证不丢失已经commit的消息的前提下容忍f个Replica的失败。在大多数使用场景中,这种模式是非常有利的。事实上,为了容忍f个Replica的失败,Majority Vote和ISR在commit前需要等待的Replica数量是一样的,但是ISR需要的总的Replica的个数几乎是Majority Vote的一半。Majority Vote与ISR相比有不需等待最慢的Broker这一优势。
     
    6.6 如何处理所有Replica都不工作
    在ISR中至少有一个follower时,Kafka可以确保已经commit的数据不丢失,但如果某个Partition的所有Replica都宕机了,Kafka会选择第一个“活”过来的Replica(不一定是ISR中的)作为Leader。
     
    6.7 如何选举Leader
    (1)方案一:所有Follower都在Zookeeper上设置一个Watch,一旦Leader宕机,其对应的ephemeral znode会自动删除,此时所有Follower都尝试创建该节点,而创建成功者(Zookeeper保证只有一个能创建成功)即是新的Leader,其它Replica即为Follower。
    此方案有3个缺点如下:
    ①split-brain 脑裂
    这是由Zookeeper的特性引起的,虽然Zookeeper能保证所有Watch按顺序触发,但并不能保证同一时刻所有Replica“看”到的状态是一样的,这就可能造成不同Replica的响应不一致。
    ②herd effect 羊群效应
    如果宕机的那个Broker上的Partition比较多,会造成多个Watch被触发,造成集群内大量的调整。
    ③Zookeeper负载过重
    每个Replica都要为此在Zookeeper上注册一个Watch,当集群规模增加到几千个Partition时Zookeeper负载会过重。
     
    (2)方案二:在所有broker中选出一个controller,所有Partition的Leader选举都由controller决定。controller会将Leader的改变直接通过RPC的方式(比Zookeeper Queue的方式更高效)通知需为此作出响应的Broker。同时controller也负责增删Topic以及Replica的重新分配。
     
     
    第7章 Kafka数据可靠性与一致性
    7.1 Partition Recovery机制
    每个Partition会在磁盘记录一个RecoveryPoint, 记录已经flush到磁盘的最大offset。当broker fail 重启时,会进行loadLogs。 首先会读取该Partition的RecoveryPoint,找到包含RecoveryPoint的segment及以后的segment,这些segment就是可能没有完全flush到磁盘segments。然后调用segment的recover,重新读取各个segment的msg,并重建索引。
    优点:
    ①以segment为单位管理Partition数据,方便数据生命周期的管理,删除过期数据简单。
    ②在程序崩溃重启时,加快recovery速度,只需恢复未完全flush到磁盘的segment。
    ③通过index中offset与物理偏移映射,用二分查找能快速定位msg,并且通过分多个Segment,每个index文件很小,查找速度更快。
     
    7.2 Partition Replica同步机制
    副本同步机制如下:
    ①Partition的多个replica中一个为Leader,其余为follower。
    ②Producer只与Leader交互,把数据写入到Leader中。
    ③Followers从Leader中拉取数据进行数据同步。
    ④Consumer只从Leader拉取数据。
     
    ISR:所有不落后的replica集合, 不落后有两层含义:距离上次FetchRequest的时间不大于某一个值或落后的消息数不大于某一个值, Leader失败后会从ISR中选取一个Follower做Leader。
     
    7.3 数据可靠性保证
    当Producer向Leader发送数据时,可以通过acks参数设置数据可靠性的级别。
    ①acks参数为0
    不论写入是否成功,server不需要给Producer发送Response,如果发生异常,server会终止连接,触发Producer更新meta数据。
    ②acks参数为1
    Leader写入成功后即发送Response,此种情况如果Leader fail,会丢失数据。
    ③acks参数为-1
    等待所有ISR接收到消息后再给Producer发送Response,这是最强保证。
    仅设置acks=-1也不能保证数据不丢失,当Isr列表中只有Leader时,同样有可能造成数据丢失。要保证数据不丢除了设置acks=-1,,还要保证ISR的大小大于等于2,
    具体参数设置:
    (1)request.required.acks设置为-1 等待所有ISR列表中的Replica接收到消息后采算写成功。
    (2)min.insync.replicas设置为大于等于2,保证ISR中至少有两个Replica。
    Producer要在吞吐率和数据可靠性之间做一个权衡。
     
    7.4 数据一致性保证
    一致性定义:若某条消息对Consumer可见,那么即使Leader宕机了,在新Leader上数据依然可以被读到。
    相关术语:
    HW:high watermark,取一个partition对应的ISR中最小的LEO(log end offset)作为HW。leader转换的时候,HW是安全线。
    LEO:log end offset,这个replica的log里最后一条消息的下一条消息的offset。
     
    (1)HighWaterMark简称HW::Partition的高水位,取一个partition对应的ISR中最小的LEO(log end offset)作为HW,消费者最多只能消费到HW所在的位置,另外每个replica都有highWatermark,leader和follower各自负责更新自己的highWatermark状态,highWatermark <= leader. LogEndOffset。offset的数据小于HW才被认为是commit的,等于HW并不是commit的。
    (2)对于Leader新写入的msg,Consumer不能立刻消费,Leader会等待该消息被所有ISR中的replica同步后,更新HW,此时该消息才能被Consumer消费,即Consumer最多只能消费到HW位置。
     
    这样就保证了如果Leader Broker失效,该消息仍然可以从新选举的Leader中获取。对于来自内部Broker的读取请求,没有HW的限制。同时,Follower也会维护一份自己的HW,Folloer.HW = min(Leader.HW, Follower.offset)。
     
    7.5 Kafka一致性重要概念
    (1)HW
    随着follower的拉取进度的即时变化,HW是随时在变化的。follower总是向leader请求自己已有messages的下一个offset开始的数据,因此当follower发出了一个fetch request,要求offset为A以上的数据,leader就知道了这个follower的log end offset至少为A。此时就可以统计下ISR里的所有replica的LEO是否已经大于了HW,如果是的话,就提高HW。同时,leader在fetch本地消息给follower时,也会在返回给follower的reponse里附带自己的HW。这样follower也就知道了leader处的HW(但是在实现中,follower获取的只是读leader本地log时的HW,并不能保证是最新的HW)。但是leader和follower的HW是不同步的,follower处记的HW可能会落后于leader。
    (2)leader和ISR
    在需要选举leader的场景下,leader和ISR是由controller决定的。在选出leader以后,ISR是leader决定。如果谁是leader和ISR只存在于ZK上,那么每个broker都需要在Zookeeper上监听它host的每个partition的leader和ISR的变化,这样效率比较低。如果不放在Zookeeper上,那么当controller fail以后,需要从所有broker上重新获得这些信息,考虑到这个过程中可能出现的问题,也不靠谱。所以leader和ISR的信息存在于Zookeeper上,但是在变更leader时,controller会先在Zookeeper上做出变更,然后再发送LeaderAndIsrRequest给相关的broker。这样可以在一个LeaderAndIsrRequest里包括这个broker上有变动的所有partition,即batch一批变更新信息给broker,更有效率。另外,在leader变更ISR时,会先在Zookeeper上做出变更,然后再修改本地内存中的ISR。
    (3)Hight Watermark Checkpoint
    由于HW是随时变化的,如果即时更新到Zookeeper,会带来效率的问题。而HW是如此重要,因此需要持久化,ReplicaManager就启动了单独的线程定期把所有的partition的HW的值记到文件中,即做highwatermark-checkpoint。
    (4)Epoch
    除了leader,ISR之外,再加上controller epoch、leader epoch和zookeeper version这三个版本号的共同作用,Kafka基本都保证对于leader, ISR, controller的认知在各个broker间不会出现大问题。
    ①controller epoch
    当新的controller开始工作后,旧的controller可能还在工作,这时就会有两个自认为是的controller,那么broker该听哪个的呢?cpmtroller epoch是一个整数,记在Zookeeper的/controller_epoch path的数据中,当新的controller当选后,它更新Zookeeper中的这个数据,把这个整数的值+1,并且以每个命令中都附带上controller epoch。这样broker收到一个controller的命令后,就与自己内存中保存的controller epoch比较,如果命令中的值小于内存中的值,就代表是旧的controller的命令,如果大于内存中的值,就更新内存中的controller epoch为新值,并且执行命令。
    ②leader epoch
    对于同一个controller,也存在它的LeaderAndIsrRequest以错误的顺序到达broker的可能,这样broker就可以在检查controller的epoch之后,再检查leader epoch,以确认该执行哪个命令。
    ③zkVersion
    对于在zookeeper path中存储的controller epoch, leaderAndIsr信息进行更新时,始终都得进行条件更新,以避免产生竞态。比如,在controller读取Zookeeper上的leaderAndIsr信息后,更新leaderAndIsr信息前,如果leader更改了ISR的信息,而controller以更改前的ISR进行leader选举的话,就可能会产生异常状态;或者在controller更新完leaderAndIsr之后,旧的leader又去更新zk上的这个数据,也会使集群不一致。所以,就需要zkVersion来进行条件变更。controller和replica在内存中存储上一次状态更新时读取到的zkVersion,当它依据此状态做出决定时,需要带上这个zkVersion做条件更新,以保证根据旧状态做出的更新不会生效。这种条件更新是使用的kafka.utils.ZkUtils的conditionalUpdatePersistentPath方法。
  • 相关阅读:
    [ZJOI2010]基站选址
    [SDOI2008]Sue的小球
    访问计划
    奥义商店
    codeforces 809E Surprise me!
    codeforces 888G Xor-MST
    [HAOI2015]数字串拆分
    小奇分糖果
    小奇的花园
    BZOJ4711 小奇挖矿
  • 原文地址:https://www.cnblogs.com/fhwup/p/8587657.html
Copyright © 2011-2022 走看看