丽水做网站的公司,四川网站建设哪家专业,高端医疗网站建设,网站服务器网络1、 kafka 是什么,有什么作用
2、Kafka为什么这么快
3、Kafka架构及名词解释
4、Kafka中的AR、ISR、OSR代表什么
5、HW、LEO代表什么
6、ISR收缩性
7、kafka follower如何与leader同步数据
8、Zookeeper 在 Kafka 中的作用#xff08;早期#xff09;
9、Kafka如何快…1、 kafka 是什么,有什么作用
2、Kafka为什么这么快
3、Kafka架构及名词解释
4、Kafka中的AR、ISR、OSR代表什么
5、HW、LEO代表什么
6、ISR收缩性
7、kafka follower如何与leader同步数据
8、Zookeeper 在 Kafka 中的作用早期
9、Kafka如何快速读取指定offset的消息
10、生产者发送消息有哪些模式
11、发送消息的分区策略有哪些
12、Kafka可靠性保证不丢消息
13、Kafka 是怎么去实现负载均衡的
14、简述Kafka的Rebalance机制
15、Kafka 负载均衡会导致什么问题
16、如何增强消费者的消费能力
17、消费者与Topic的分区策略
18、如何保证消息不被重复消费消费者幂等性
19、为什么Kafka不支持读写分离
20、Kafka选举机制
21、脑裂问题
22、如何为Kafka集群选择合适的Topics/Partitions数量
23、Kafka 分区数可以增加或减少吗为什么
24、谈谈你对Kafka生产者幂等性的了解
25、谈谈你对 Kafka事务的了解
26、Kafka消息是采用Pull模式还是Push模式
27、Kafka缺点
28、Kafka什么时候会丢数据
29、Kafka分区数越多越好吗
30、Kafka如何保证消息的有序性
31、Kafka精确一次性Exactly-once如何保证
1、 kafka 是什么,有什么作用
Kafka是一个开源的高吞吐量的分布式消息中间件对比于其他 1) 缓冲和削峰上游数据时有突发流量下游可能扛不住或者下游没有足够多的机器来保证冗余kafka在中间可以起到一个缓冲的作用把消息暂存在kafka中下游服务就可以按照自己的节奏进行慢慢处理。
1) 解耦和扩展性项目开始的时候并不能确定具体需求。消息队列可以作为一个接口层解耦重要的业务流程。只需要遵守约定针对数据编程即可获取扩展能力。
1) 冗余可以采用一对多的方式一个生产者发布消息可以被多个订阅topic的服务消费到供多个毫无关联的业务使用。
1) 健壮性消息队列可以堆积请求所以消费端业务即使短时间死掉也不会影响主要业务的正常进行。
1) 异步通信很多时候用户不想也不需要立即处理消息。消息队列提供了异步处理机制允许用户把一个消息放入队列但并不立即处理它。想向队列中放入多少消息就放多少然后在需要的时候再去处理它们。
2、Kafka为什么这么快
利用 Partition 实现并行处理 不同 Partition 可位于不同机器因此可以充分利用集群优势实现机器间的并行处理。另一方面由于 Partition 在物理上对应一个文件夹即使多个 Partition 位于同一个节点也可通过配置让同一节点上的不同 Partition 置于不同的磁盘上从而实现磁盘间的并行处理充分发挥多磁盘的优势。利用了现代操作系统分页存储 Page Cache 来利用内存提高 I/O 效率顺序写 kafka的消息是不断追加到文件中的这个特性使kafka可以充分利用磁盘的顺序读写性能 由于现代的操作系统提供了预读和写技术磁盘的顺序写大多数情况下比随机写内存还要快。顺序读写不需要硬盘磁头的寻道时间只需很少的扇区旋转时间所以速度远快于随机读写Zero-copy 零拷技术减少拷贝次数数据批量处理。合并小的请求然后以流的方式进行交互直顶网络上限。在很多情况下系统的瓶颈不是 CPU 或磁盘而是网络IO。因此除了操作系统提供的低级批处理之外Kafka 的客户端和 broker 还会在通过网络发送数据之前在一个批处理中累积多条记录 (包括读和写)。记录的批处理分摊了网络往返的开销使用了更大的数据包从而提高了带宽利用率。Pull 拉模式 使用拉模式进行消息的获取消费与消费端处理能力相符。数据压缩 Kafka还支持对消息集合进行压缩Producer可以通过GZIP、Snappy、LZ4格式对消息集合进行压缩数据压缩一般都是和批处理配套使用来作为优化手段的。压缩的好处就是减少传输的数据量减轻对网络传输的压力 Producer压缩之后在Consumer需进行解压虽然增加了CPU的工作但在对大数据处理上瓶颈在网络上而不是CPU所以这个成本很值得
3、Kafka架构及名词解释
简易架构图如下 详细架构图如下
Broker 一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。Producer消息生产者向kafka broker发送消息的客户端。Consumer消息消费者向kafka broker取消息的客户端。Topic队列生产者和消费者通过此进行对接。Consumer Group CG若干个Consumer组成的集合。这是kafka用来实现一个topic消息的广播发给所有的consumer和单播发给任意一个consumer的手段。一个topic可以有多个CG。topic的消息会复制不是真的复制是概念上的到所有的CG但每个CG只会把消息发给该CG中的一个consumer。如果需要实现广播只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。Partition分区为了实现扩展性一个topic可以分布在多个broker上一个topic可以分为多个partition每个partition都是一个有序的队列。partition中的每条消息都会被分配一个有序的idoffset。kafka只保证同一个partition中的消息顺序不保证一个topic的整体多个partition之间的顺序。生产者和消费者使用时可以指定topic中的具体partition。副本在kafka中每个主题可以有多个分区每个分区又可以有多个副本。这多个副本中只有一个是leader而其他的都是follower副本。仅有leader副本可以对外提供服务。多个follower副本通常存放在和leader副本不同的broker中。通过这样的机制实现了高可用当某台机器挂掉后其他follower副本也能迅速”转正“开始对外提供服务。offset消费偏移量topic中的每个分区都是有序且顺序不可变的记录集并且不断地追加到结构化的log文件。分区中的每一个记录都会分配一个id号来表示顺序我们称之为offsetoffset用来唯一的标识分区中每一条记录。可以设置为“自动提交”与“手动提交”。
4、Kafka中的AR、ISR、OSR代表什么
AR:Assigned Replicas 指当前分区中的所有副本。ISR:In-Sync Replicas 副本同步队列。ISR中包括Leader和Foller。如果Leader进程挂掉会在ISR队列中选择一个服务作为新的Leader。有replica.lag.max.message(延迟条数)和replica.lag.time.max.ms(延迟时间)两个参数决定一台服务器是否可以加入ISR副本队列在0.10版本之后移除了replica.lag.max.message(延迟条数)参数防治服务频繁的进出队列。任意一个维度超过阈值都会把Follower踢出ISR存入OSROutof-Sync Replicas列表新加入的Follower也会先存放在OSR中。OSROut-of-Sync Replicas非同步副本队列。与leader副本同步滞后过多的副本不包括leader副本组成OSR。如果OSR集合中有follower副本“追上”了leader副本那么leader副本会把它从OSR集合转移至ISR集合。默认情况下当leader副本发生故障时只有在ISR集合中的副本才有资格被选举为新的leader而在OSR集合中的副本则没有任何机会不过这个原则也可以通过修改unclean.leader.election.enable参数配置来改变。unclean.leader.election.enable 为true的话意味着非ISR集合的broker 也可以参与选举这样就有可能发生数据丢失和数据不一致的情况Kafka的可靠性就会降低而如果unclean.leader.election.enable参数设置为falseKafka的可用性就会降低。 ISR的伸缩1Leader跟踪维护ISR中follower滞后状态落后太多或失效时leade把他们从ISR剔除。2OSR中follower“追上”Leader在ISR中才有资格选举leader。 5、HW、LEO代表什么 LEO Log End Offset标识当前日志文件中下一条待写入的消息的offset。上图中offset为9的位置即为当前日志文件的 LEOLEO 的大小相当于当前日志分区中最后一条消息的offset值加1.分区 ISR 集合中的每个副本都会维护自身的 LEO 而 ISR 集合中最小的 LEO 即为分区的 HW对消费者而言只能消费 HW 之前的消息。HWreplica高水印值副本中最新一条已提交消息的位移。leader 的HW值也就是实际已提交消息的范围每个replica都有HW值但仅仅leader中的HW才能作为标示信息。什么意思呢就是说当按照参数标准成功完成消息备份成功同步给follower replica后才会更新HW的值代表消息理论上已经不会丢失可以认为“已提交”。 6、ISR收缩性
启动 Kafka时候自动开启的两个定时任务“isr-expiration和”isr-change-propagation。
isr-expirationisr-expiration任务会周期性的检测每个分区是否需要缩减其ISR集合相当于一个纪检委员巡查尖子班时候发现有学生睡觉打牌看小说就把它的座位移除尖子班缩减ISR宁缺毋滥。同样道理如果follower数据同步赶上leader那么该follower就能进入ISR尖子班扩充。上面关于ISR尖子班人员的所见都会记录到isrChangeSet中想象成是一个名单列表谁能进谁要出都记录在案。isr-change-propagation作用就是检查isrChangeSet按照名单上的信息移除和迁入一般是2500ms检查一次但是为了防止频繁收缩扩充影响性能不是每次都能做变动必须满足1、上一次ISR集合发生变化距离现在已经超过5秒2、上一次写入zookeeper的时候距离现在已经超过60秒。这两个条件都满足那么就开始换座位这两个条件可以由我们来配置。Kafka使用这种ISR收缩的方式有效的权衡了数据可靠性与性能之间的关系。
7、kafka follower如何与leader同步数据
Kafka的复制机制既不是完全的同步复制也不是单纯的异步复制。完全同步复制要求All Alive Follower都复制完这条消息才会被认为commit这种复制方式极大的影响了吞吐率。而异步复制方式下Follower异步的从Leader复制数据数据只要被Leader写入log就被认为已经commit这种情况下如果leader挂掉会丢失数据kafka使用ISR的方式很好的均衡了确保数据不丢失以及吞吐率。Follower可以批量的从Leader复制数据而且Leader充分利用磁盘顺序读以及send file(zero copy)机制这样极大的提高复制性能内部批量写磁盘大幅减少了Follower与Leader的消息量差。
8、Zookeeper 在 Kafka 中的作用早期
zookeeper 是一个分布式的协调组件早期版本的kafka用zk做meta信息存储consumer的消费状态group的管理以及 offset的值。考虑到zk本身的一些因素以及整个架构较大概率存在单点问题新版本中逐渐弱化了zookeeper的作用。新的consumer使用了kafka内部的group coordination协议也减少了对zookeeper的依赖
但是broker依然依赖于ZKzookeeper 在kafka中还用来选举controller 和 检测broker是否存活等等。
1. Broker注册Broker是分布式部署并且互相独立此时需要有一个注册系统能够将整个集群中的Broker管理起来此时就用到的Zookeeper。在Zookeeper上会有一个专门用来进行Broker服务器列表记录的节点/brokes/ids
2.Topic注册在kafka中同一个Topic的消息会被分成多个分区并将其分布在多个Broker上这些分区信息以及与Broker的对应关系也都是由Zookeeper维护由专门的节点记录/brokers/topics
3.消费者注册消费者服务器在初始化启动时加入消费者分组的步骤如下注册到消费者分组。每个消费者服务器启动时都会到Zookeeper的指定节点下创建一个属于自己的消费者节点例如/consumer/[groupid]/ids/[consumerid]完成节点创建后消费者就会将自己订阅的Topic信息写入该临时节点。
对消费者分组中的消费者的变化注册监听每个消费者都需要关注所属消费者分组中的其他消费者服务器的变化情况即对/consumer/[group_id]/ids节点注册子节点变化的Watcher监听一旦发现消费者新增或减少就触发消费者的负载均衡。对Broker服务器变化注册监听消费者需要对/broker/ids[0-N]中的节点进行监听如果发现Broker服务器列表发生变化那么就根据具体情况来决定是否需要进行消费者负载均衡。进行消费者负载均衡为了让同一个Topic下不同分区的消息尽量均衡地被多个消费者消费而进行消费者与消息分区分配的过程通常对于一个消费者分组如果组内的消费者服务器发生变更或Broker服务器发生变更会进行消费者负载均衡。Offset记录 在消费者对指定消息分区进行消费的过程中需要定时地将分区消息的消费进度Offset记录到Zookeeper上以便对该消费者进行重启或者其他消费者重新接管该消息分区的消息消费后能够从之前的进度继续进行消息消费。Offset在Zookeeper中由一个专门节点进行记录其节点路径为/consumers/[groupid]/offsets/[topic]/[brokerid-partition_id] 节点内容就是Offset的值。
4.生产者负载均衡由于同一个Topic消息会被分区并将其分布在多个Broker上因此生产者需要将消息合理地发送到这些分布式的Broker上那么如何实现生产者的负载均衡Kafka支持传统的四层负载均衡也支持Zookeeper方式实现负载均衡。
四层负载均衡根据生产者的IP地址和端口来为其圈定一个相关联的Broker。通常一个生产者只会对应单个Broker然后该生产者产生的消息都发送到该Broker。这种方式逻辑简单每个生产者不需要同其他系统建立额外的TCP链接只需要和Broker维护单个TCP连接即可。但是无法做到真正的负载均衡因为实际系统中的每个生产者产生的消息量及每个Broker的消息存储量都是不一样的如果有些生产者产生的消息远多于其他生产者的话那么会导致不同的Broker接收到的消息总数差异巨大同时生产者也无法实时感知到Broker的新增和删除。使用Zookeeper进行负载均衡由于每个Broker启动时都会完成Broker注册过程生产者会通过该节点的变化来动态地感知到Broker服务器列表的变更这样就可以实现动态的负载均衡机制。
5.消费者负载均衡与生产者相似Kafka中的消费者同样需要进行负载均衡来实现多个消费者合理地从对应的Broker服务器上接收消息每个消费者分组包含若干消费者每条消息都只会发送给分组中的一个消费者不同的消费者分组消费自己特定的Topic下面的消息互不干扰。
6.分区与消费者的关系消费组consumer group下有多个Consumer消费者。对于每个消费者组consumer groupKafka都会为其分配一个全局唯一的Group IDGroup内部的所有消费者共享该ID。订阅的topic下的每个分区只能分配给某个group下的一个consumer当然该分区还可以被分配给其他group 同时kafka为每个消费者分配一个Consumer ID通常采用“HostnameUUID”形式表示。在kafka中规定了每个消息分区只能被同组的一个消费者进行消费因此需要在zookeeper上记录消息分区与Consumer之间的关系每个消费者一旦确定了对一个消费分区的消费权利需要将其Consumer ID写入到平Zookeeper对应消息分区的临时节点上例如/consumers/[groupid]/owners/topic/[brokerid-partitionid] 其中[brokerid-partition_id]就是一个消息分区的表示节点内容就是该消息分区上消费者的Consumer ID。
7.补充早期版本的 kafka 用 zk 做 meta 信息存储consumer 的消费状态group 的管理以及 offse t的值。考虑到zk本身的一些因素以及整个架构较大概率存在单点问题新版本中确实逐渐弱化了zookeeper的作用。新的consumer使用了kafka内部的group coordination协议也减少了对zookeeper的依赖
9、Kafka如何快速读取指定offset的消息 Kafka本地日志存储根据segement分段存储默认1G其中segement包括index稀疏索引文件和log数据文件。其中index文件索引通过offset与posttion来定位数据文件中指定message的消息。其中index和log的文件名都为当前segement的起始offset。
读取offset170418的消息首先通过offset根据二分法定位到index索引文件然后根据索引文件中的[offset,position]position为物理偏移地址去log中获取指定offset的message数据。
10、生产者发送消息有哪些模式
异步发送
对于生产者的异步发送来说就是我发送完当前消息后并不需要你将当前消息的发送结果立马告诉我而是可以随即进行下一条消息的发送。但是我会允许添加一个回调函数接收你后续返回的发送结果。异步发送这块我们直接调用kafkaProducer的send方法即可实现异步发送。
同步发送
如果生产者需要使用同步发送的方式只需要拿到 send 方法返回的future对象后调用其 get() 方法即可。此时如果消息还未发送到broker中get方法会被阻塞等到 broker 返回消息发送结果后会跳出当前方法并将结果返回。
11、发送消息的分区策略有哪些
所谓分区写入策略即是生产者将数据写入到kafka主题后kafka如何将数据分配到不同分区中的策略。
常见的有三种策略轮询策略随机策略和按键保存策略。其中轮询策略是默认的分区策略而随机策略则是较老版本的分区策略不过由于其分配的均衡性不如轮询策略故而后来改成了轮询策略为默认策略。
轮询策略
所谓轮询策略即按顺序轮流将每条数据分配到每个分区中。
举个例子假设主题test有三个分区分别是分区A分区B和分区C。那么主题对接收到的第一条消息写入A分区第二条消息写入B分区第三条消息写入C分区第四条消息则又写入A分区依此类推。
轮询策略是默认的策略故而也是使用最频繁的策略它能最大限度保证所有消息都平均分配到每一个分区。除非有特殊的业务需求否则使用这种方式即可。 随机策略
随机策略也就是每次都随机地将消息分配到每个分区。其实大概就是先得出分区的数量然后每次获取一个随机数用该随机数确定消息发送到哪个分区。
在比较早的版本默认的分区策略就是随机策略但其实使用随机策略也是为了更好得将消息均衡写入每个分区。但后来发现对这一需求而言轮询策略的表现更优所以社区后来的默认策略就是轮询策略了。 hashKey
按键保存策略就是当生产者发送数据的时候可以指定一个key计算这个key的hashCode值按照hashCode的值对不同消息进行存储。
至于要如何实现那也简单只要让生产者发送的时候指定key就行。欸刚刚不是说默认的是轮询策略吗其实啊kafka默认是实现了两个策略没指定key的时候就是轮询策略有的话那激素按键保存策略了。
上面有说到一个场景那就是要顺序发送消息到kafka。前面提到的方案是让所有数据存储到一个分区中但其实更好的做法就是使用这种按键保存策略。
让需要顺序存储的数据都指定相同的键而不需要顺序存储的数据指定不同的键这样一来即实现了顺序存储的需求又能够享受到kafka多分区的优势岂不美哉。 粘性分区
所以如果使用默认的轮询partition策略可能会造成一个大的batch被轮询成多个小的batch的情况。鉴于此kafka2.4的时候推出一种新的分区策略即StickyPartitioning StrategyStickyPartitioning Strategy会随机地选择另一个分区并会尽可能地坚持使用该分区——即所谓的粘住这个分区。
鉴于小batch可能导致延时增加之前对于无Key消息的分区策略效率很低。社区于2.4版本引入了黏性分区策略StickyPartitioning Strategy。该策略是一种全新的策略能够显著地降低给消息指定分区过程中的延时。使用StickyPartitioner有助于改进消息批处理减少延迟并减少broker的负载。
自定义分区器
实现partitioner接口 切记分区是实现负载均衡以及高吞吐量的关键所以一定要在生产者这一端就要考虑好合适的分区策略避免造成消息数据的“倾斜”使得某些分区成为性能瓶颈从而导致下游数据消费的性能下降的问题。 12、Kafka可靠性保证不丢消息 Kafka精确一次性Exactly-once保障之一 Kafka可靠性主要从三个方面来看Broker、Producer、Consumer。1. Brokerbroker写数据时首先写到PageCache中pageCache的数据通过linux的flusher程序异步批量存储至磁盘中此过程称为刷盘。而pageCache位于内存。这部分数据会在断电后丢失。刷盘触发条件有三
主动调用sync或fsync函数可用内存低于阀值dirty data时间达到阀值。dirty是pagecache的一个标识位当有数据写入到pageCache时pagecache被标注为dirty数据刷盘以后dirty标志清除。
kafka没有提供同步刷盘的方式也就是说理论上要完全让kafka保证单个broker不丢失消息是做不到的只能通过调整刷盘机制的参数缓解该情况比如 减少刷盘间隔log.flush.interval.ms(在刷新到磁盘之前任何topic中的消息保留在内存中的最长时间) 减少刷盘数据量大小log.flush.interval.messages(在将消息刷新到磁盘之前在日志分区上累积的消息数量)。 时间越短数据量越小性能越差但是丢失的数据会变少可靠性越好。这是一个选择题。
同时Kafka通过producer和broker协同处理消息丢失的情况一旦producer发现broker消息丢失即可自动进行retry。retry次数可根据参数retries进行配置超过指定次数会此条消息才会被判断丢失。producer和broker之间通过ack机制来判断消息是否丢失。
acks0producer不等待broker的响应效率最高但是消息很可能会丢。acks1leader broker收到消息后不等待其他follower的响应即返回ack。也可以理解为ack数为1。此时如果follower还没有收到leader同步的消息leader就挂了那么消息会丢失。按照上图中的例子如果leader收到消息成功写入PageCache后会返回ack此时producer认为消息发送成功。但此时按照上图数据还没有被同步到follower。如果此时leader断电数据会丢失。acks-1leader broker收到消息后挂起等待所有ISR列表中的follower返回结果后再返回ack。-1等效与all。这种配置下只有leader写入数据到pagecache是不会返回ack的还需要所有的ISR返回“成功”才会触发ack。如果此时断电producer可以知道消息没有被发送成功将会重新发送。如果在follower收到数据以后成功返回ackleader断电数据将存在于原来的follower中。在重新选举以后新的leader会持有该部分数据。数据从leader同步到follower需要2步数据从pageCache被刷盘到disk。因为只有disk中的数据才能被同步到replica。数据同步到replica并且replica成功将数据写入PageCache。在producer得到ack后哪怕是所有机器都停电数据也至少会存在于leader的磁盘内。上面第三点提到了ISR的列表的follower需要配合另一个参数才能更好的保证ack的有效性。ISR是Broker维护的一个“可靠的follower列表”in-sync Replica列表broker的配置包含一个参数min.insync.replicas。该参数表示ISR中最少的副本数。如果不设置该值ISR中的follower列表可能为空。此时相当于acks1。
Topic 分区副本
在 Kafka 0.8.0 之前Kafka 是没有副本的概念的那时候人们只会用 Kafka 存储一些不重要的数据因为没有副本数据很可能会丢失。但是随着业务的发展支持副本的功能越来越强烈所以为了保证数据的可靠性Kafka 从 0.8.0 版本开始引入了分区副本详情请参见 KAFKA-50。也就是说每个分区可以人为的配置几个副本比如创建主题的时候指定 replication-factor也可以在 Broker 级别进行配置 default.replication.factor一般会设置为3。
Kafka 可以保证单个分区里的事件是有序的分区可以在线可用也可以离线不可用。在众多的分区副本里面有一个副本是 Leader其余的副本是 follower所有的读写操作都是经过 Leader 进行的同时 follower 会定期地去 leader 上的复制数据。当 Leader 挂了的时候其中一个 follower 会重新成为新的 Leader。通过分区副本引入了数据冗余同时也提供了 Kafka 的数据可靠性。
Kafka 的分区多副本架构是 Kafka 可靠性保证的核心把消息写入多个副本可以使 Kafka 在发生崩溃时仍能保证消息的持久性。
2. Producer
producer在发送数据时可以将多个请求进行合并后异步发送合并后的请求首先缓存在本地buffer中正常情况下producer客户端的异步调用可以通过callback回调函数来处理消息发送失败或者超时的情况但是当出现以下情况将会出现数据丢失
producer异常中断buffer中的数据将丢失。producer客户端内存不足如果采取的策略是丢弃消息另一种策略是block阻塞消息也会丢失。消息产生异步过快导致挂起线程过多内存不足导致程序崩溃消息丢失。
针对以上情况可以有以下解决思路。
producer采用同步方式发送消息或者生产数据时采用阻塞的线程池并且线程数不宜过多。整体思路就是控制消息产生速度。扩大buffer的容量配置配置项为buffer.memory。这种方法可以缓解数据丢失的情况但不能杜绝。
3.Consumer
Consumer消费消息有以下几个步骤
接收消息处理消息反馈处理结果
消费方式主要分为两种
自动提交offsetAutomatic Offset Committing enable.auto.committrue手动提交offsetManual Offset Controlenable.auto.commitfalse
Consumer自动提交机制是根据一定的时间间隔将收到的消息进行commit具体配置为auto.commit.interval.ms。commit和消费的过程是异步的也就是说可能存在消费过程未成功commit消息就已经提交此时就会出现消息丢失。我们可将提交类型改为手动提交在消费完成后再进行提交这样可以保证消息“至少被消费一次”at least once但如果消费完成后在提交过程中出现故障则会出现重复消费的情况本章不讨论下章讲解。
13、Kafka 是怎么去实现负载均衡的
生产者层面
分区器是生产者层面的负载均衡。Kafka 生产者生产消息时根据分区器将消息投递到指定的分区中所以 Kafka 的负载均衡很大程度上依赖于分区器。Kafka 默认的分区器是 Kafka 提供的 DefaultPartitioner。它的分区策略是根据 Key 值进行分区分配的
如果 key 不为 null对 Key 值进行 Hash 计算从所有分区中根据 Key 的 Hash 值计算出一个分区号拥有相同 Key 值的消息被写入同一个分区如果 key 为 null消息将以轮询的方式在所有可用分区中分别写入消息。如果不想使用 Kafka 默认的分区器用户可以实现 Partitioner 接口自行实现分区方法。 注在笔者的理解中分区器的负载均衡与顺序性有着一定程度上的矛盾。 负载均衡的目的是将消息尽可能平均分配对于 Kafka 而言就是尽可能将消息平均分配给所有分区如果使用 Kafka 保证顺序性则需要利用到 Kafka 的分区顺序性的特性。对于需要保证顺序性的场景通常会利用 Key 值实现分区顺序性那么所有 Key值相同的消息就会进入同一个分区。这样的情况下对于大量拥有相同 Key值的消息会涌入同一个分区导致一个分区消息过多其他分区没有消息的情况即与负载均衡的思想相悖。
消费者层面
主要根据消费者的Rebalance机制实现内容详见下章
14、简述Kafka的Rebalance机制
什么是 Rebalance Rebalance 本质上是一种协议规定了一个 Consumer Group 下的所有 consumer 如何达成一致来分配订阅 Topic 的每个分区。 例如某 Group 下有 20 个 consumer 实例它订阅了一个具有 100 个 partition 的 Topic。正常情况下kafka 会为每个 Consumer 平均的分配 5 个分区。这个分配的过程就是 Rebalance。 触发 Rebalance 的时机 Rebalance 的触发条件有3个。 组成员个数发生变化。例如有新的 consumer 实例加入该消费组或者离开组。订阅的 Topic 个数发生变化。订阅 Topic 的分区数发生变化。
Rebalance 发生时Group 下所有 consumer 实例都会协调在一起共同参与kafka 能够保证尽量达到最公平的分配。但是 Rebalance 过程对 consumer group 会造成比较严重的影响。在 Rebalance 的过程中 consumer group 下的所有消费者实例都会停止工作等待 Rebalance 过程完成。
Rebalance 过程 Rebalance 过程分为两步JoinGroup 请求和 SyncGroup 请求。JoinGroup :JoinGroup 请求的主要作用是将组成员订阅信息发送给领导者消费者待领导者制定好分配方案后重平衡流程进入到 SyncGroup 请求阶段。SyncGroupSyncGroup 请求的主要目的就是让协调者把领导者制定的分配方案下发给各个组内成员。当所有成员都成功接收到分配方案后消费者组进入到 Stable 状态即开始正常的消费工作。 15、Kafka 负载均衡会导致什么问题
在消费者组Rebalance期间一直等到rebalance结束前消费者会出现无法读取消息造成整个消费者组一段时间内不可用。
16、如何增强消费者的消费能力
1、如果是Kafka消费能力不足则可以考虑增加Topic的分区数并且同时提升消费组的消费者数量消费者数分区数。两者缺一不可。
2、如果是下游的数据处理不及时则提高每批次拉取的数量。批次拉取数据过少拉取数据/处理时间生产速度使处理的数据小于生产的数据也会造成数据积压。
3、优化消费者的处理逻辑提高处理效率
17、消费者与Topic的分区策略
Range
Range是对每个Topic而言的即一个Topic一个Topic分首先对同一个Topic里面的分区按照序号进行排序并对消费者按照字母顺序进行排序。然后用Partitions分区的个数除以消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽那么前面几个消费者线程将会多消费一个分区。
RoundRobin
将消费组内所有消费者以及消费者所订阅的所有topic的partition按照字典序排序然后通过轮询方式逐个将分区以此分配给每个消费者。使用RoundRobin策略有两个前提条件必须满足
同一个消费者组里面的所有消费者的num.streams消费者消费线程数必须相等每个消费者订阅的主题必须相同。
StickyAssignor
无论是RangeAssignor还是RoundRobinAssignor当前的分区分配算法都没有考虑上一次的分配结果。显然在执行一次新的分配之前如果能考虑到上一次分配的结果尽量少的调整分区分配的变动显然是能节省很多开销的。
Sticky是“粘性的”可以理解为分配结果是带“粘性的”——每一次分配变更相对上一次分配做最少的变动上一次的结果是有粘性的其目标有两点
分区的分配尽量的均衡每一次重分配的结果尽量与上一次分配结果保持一致
StickyAssignor的模式比其他两种提供更加均衡的分配结果在发生Consumer或者Partition变更的情况下也能减少不必要的分区调整。
18、如何保证消息不被重复消费消费者幂等性 Kafka精确一次性Exactly-once保障之一 幂等性就是用户对于同一操作发起的一次请求或者多次请求的结果是一致的不会因为多次点击而产生了副作用。
出现原因
原因1Consumer在消费过程中被强行kill掉消费者线程或异常中断消费系统宕机、重启等导致实际消费后的数据offset没有提交。原因2设置offset为自动提交关闭kafka时如果在close之前调用 consumer.unsubscribe() 则有可能部分offset没提交下次重启会重复消费。原因3消费超时导致消费者与集群断开连接offset尚未提交导致重平衡后重复消费。一般消费超时session.time.out有以下原因并发过大消费者突然宕机处理超时等。
解决思路
提高消费能力提高单条消息的处理速度例如对消息处理中比 较耗时的步骤可通过异步的方式进行处理、利用多线程处理等。在缩短单条消息消费时常的同时根据实际场景可将session.time.outConsumer心跳超时时间和max.poll.interval.msconsumer两次poll的最大时间间隔值设置大一点避免不必要的rebalance此外可适当减小max.poll.records的值 表示每次消费的时候获取多少条消息默认值是500可根据实际消息速率适当调小。这种思路可解决因消费时间过长导致的重复消费问题 对代码改动较小但无法绝对避免重复消费问题。根据业务情况制定引入单独去重机制例如生成消息时在消息中加入唯一标识符如主键id。写入时根据逐渐主键判断update还是insert。如果写redis则每次根据主键id进行set即可天然幂等性。或者使用redis作为缓冲将id首先写入redis进行重复判断然后在进行后续操作。开启生产者的精确一次性也就是幂等性 再引入producer事务 即客户端传入一个全局唯一的Transaction ID这样即使本次会话挂掉也能根据这个id找到原来的事务状态
19、为什么Kafka不支持读写分离
在 Kafka 中生产者写入消息、消费者读取消息的操作都是与 leader 副本进行交互的从 而实现的是一种主写主读的生产消费模型。
Kafka 并不支持主写从读因为主写从读有 2 个很明 显的缺点:
数据一致性问题。数据从主节点转到从节点必然会有一个延时的时间窗口这个时间 窗口会导致主从节点之间的数据不一致。某一时刻在主节点和从节点中 A 数据的值都为 X 之后将主节点中 A 的值修改为 Y那么在这个变更通知到从节点之前应用读取从节点中的 A 数据的值并不为最新的 Y由此便产生了数据不一致的问题。延时问题。类似 Redis 这种组件数据从写入主节点到同步至从节点中的过程需要经 历网络→主节点内存→网络→从节点内存这几个阶段整个过程会耗费一定的时间。而在 Kafka 中主从同步会比 Redis 更加耗时它需要经历网络→主节点内存→主节点磁盘→网络→从节 点内存→从节点磁盘这几个阶段。对延时敏感的应用而言主写从读的功能并不太适用。
20、Kafka选举机制
Kafka选举主要分为以下三种
控制器Broker选举机制分区副本选举机制消费组选举机制
控制器选举
控制器是Kafka的核心组件它的主要作用是在Zookeeper的帮助下管理和协调整个Kafka集群包括所有分区与副本的状态。集群中任意一个Broker都能充当控制器的角色但在运行过程中只能有一个Broker成为控制器。集群中第一个启动的Broker会通过在Zookeeper中创建临时节点/controller来让自己成为控制器其他Broker启动时也会在zookeeper中创建临时节点但是发现节点已经存在所以它们会收到一个异常意识到控制器已经存在那么就会在Zookeeper中创建watch对象便于它们收到控制器变更的通知。如果控制器与Zookeeper断开连接或异常退出其他broker通过watch收到控制器变更的通知就会尝试创建临时节点/controller如果有一个Broker创建成功那么其他broker就会收到创建异常通知代表控制器已经选举成功其他Broker只需创建watch对象即可。
控制器作用
主题管理创建、删除Topic以及增加Topic分区等操作都是由控制器执行。分区重分配执行Kafka的reassign脚本对Topic分区重分配的操作也是由控制器实现。如果集群中有一个Broker异常退出控制器会检查这个broker是否有分区的副本leader如果有那么这个分区就需要一个新的leader此时控制器就会去遍历其他副本决定哪一个成为新的leader同时更新分区的ISR集合。如果有一个Broker加入集群中那么控制器就会通过Broker ID去判断新加入的Broker中是否含有现有分区的副本如果有就会从分区副本中去同步数据。Preferred leader选举因为在Kafka集群长时间运行中broker的宕机或崩溃是不可避免的leader就会发生转移即使broker重新回来也不会是leader了。在众多leader的转移过程中就会产生leader不均衡现象可能一小部分broker上有大量的leader影响了整个集群的性能所以就需要把leader调整回最初的broker上这就需要Preferred leader选举。集群成员管理控制器能够监控新broker的增加broker的主动关闭与被动宕机进而做其他工作。这也是利用Zookeeper的ZNode模型和Watcher机制控制器会监听Zookeeper中/brokers/ids下临时节点的变化。同时对broker中的leader节点进行调整。元数据服务控制器上保存了最全的集群元数据信息其他所有broker会定期接收控制器发来的元数据更新请求从而更新其内存中的缓存数据。
分区副本选举机制
发生副本选举的情况
创建主题增加分区分区下线分区中原先的leader副本下线此时分区需要选举一个新的leader上线来对外提供服务分区重分配
分区leader副本的选举由Kafka控制器负责具体实施。主要过程如下
从Zookeeper中读取当前分区的所有ISR(in-sync replicas)集合。调用配置的分区选择算法选择分区的leader。
分区副本分为ISR同步副本和OSR非同步副本当leader发生故障时只有“同步副本”才可以被选举为leader。选举时按照集合中副本的顺序查找第一个存活的副本并且这个副本在ISR集合中。同时kafka支持OSR非同步副本也参加选举Kafka broker端提供了一个参数unclean.leader.election.enable用于控制是否允许非同步副本参与leader选举如果开启则当 ISR为空时就会从这些副本中选举新的leader这个过程称为 Unclean leader选举。可以根据实际的业务场景选择是否开启Unclean leader选举。开启 Unclean 领导者选举可能会造成数据丢失但好处是它使得分区 Leader 副本一直存在不至于停止对外提供服务因此提升了高可用性。一般建议是关闭Unclean leader选举因为通常数据的一致性要比可用性重要。
消费组Consumer Group选主
在Kafka的消费端会有一个消费者协调器以及消费组组协调器Group Coordinator需要为消费组内的消费者选举出一个消费组的leader。如果消费组内还没有leader那么第一个加入消费组的消费者即为消费组的leader如果某一个时刻leader消费者由于某些原因退出了消费组那么就会重新选举leader选举源码如下
private val members new mutable.HashMap[String, MemberMetadata]
leaderId members.keys.headOption
在组协调器中消费者的信息是以HashMap的形式存储的其中key为消费者的member_id而value是消费者相关的元数据信息。而leader的取值为HashMap中的第一个键值对的key这种选举方式等同于随机。
消费组的Leader和Coordinator没有关联。消费组的leader负责Rebalance过程中消费分配方案的制定。
21、脑裂问题
controller挂掉后Kafka集群会重新选举一个新的controller。这里面存在一个问题很难确定之前的controller节点是挂掉还是只是短暂性的故障。如果之前挂掉的controller又正常了他并不知道自己已经被取代了那么此时集群中会出现两台controller。
其实这种情况是很容易发生。比如某个controller由于GC而被认为已经挂掉并选择了一个新的controller。在GC的情况下在最初的controller眼中并没有改变任何东西该Broker甚至不知道它已经暂停了。因此它将继续充当当前controller这是分布式系统中的常见情况称为脑裂。
假如处于活跃状态的controller进入了长时间的GC暂停。它的ZooKeeper会话过期了之前注册的/controller节点被删除。集群中其他Broker会收到zookeeper的这一通知。 由于集群中必须存在一个controller Broker所以现在每个Broker都试图尝试成为新的controller。假设Broker 2速度比较快成为了最新的controller Broker。此时每个Broker会收到Broker2成为新的controller的通知由于Broker3正在进行stop the world的GC可能不会收到Broker2成为最新的controller的通知。 等到Broker3的GC完成之后仍会认为自己是集群的controller在Broker3的眼中好像什么都没有发生一样。 现在集群中出现了两个controller它们可能一起发出具有冲突的命令就会出现脑裂的现象。如果对这种情况不加以处理可能会导致严重的不一致。所以需要一种方法来区分谁是集群当前最新的Controller。
Kafka是通过使用epoch number纪元编号也称为隔离令牌来完成的。epoch number只是单调递增的数字第一次选出Controller时epoch number值为1如果再次选出新的Controller则epoch number将为2依次单调递增。
每个新选出的controller通过Zookeeper 的条件递增操作获得一个全新的、数值更大的epoch number 。其他Broker 在知道当前epoch number 后如果收到由controller发出的包含较旧(较小)epoch number的消息就会忽略它们即Broker根据最大的epoch number来区分当前最新的controller。 上图Broker3向Broker1发出命令:让Broker1上的某个分区副本成为leader该消息的epoch number值为1。于此同时Broker2也向Broker1发送了相同的命令不同的是该消息的epoch number值为2此时Broker1只听从Broker2的命令(由于其epoch number较大)会忽略Broker3的命令从而避免脑裂的发生。
22、如何为Kafka集群选择合适的
Topics/Partitions数量
1、根据当前topic的消费者数量确认
在kafka中单个patition是kafka并行操作的最小单元。在producer和broker端向每一个分区写入数据是可以完全并行化的此时可以通过加大硬件资源的利用率来提升系统的吞吐量例如对数据进行压缩。在consumer段kafka只允许单个partition的数据被一个consumer线程消费。因此在consumer端每一个Consumer Group内部的consumer并行度完全依赖于被消费的分区数量。综上所述通常情况下在一个Kafka集群中partition的数量越多意味着可以到达的吞吐量越大。
2、根据consumer端的最大吞吐量确定
我们可以粗略地通过吞吐量来计算kafka集群的分区数量。假设对于单个partitionproducer端的可达吞吐量为pConsumer端的可达吞吐量为c期望的目标吞吐量为t那么集群所需要的partition数量至少为max(t/p,t/c)。在producer端单个分区的吞吐量大小会受到批量大小、数据压缩方法、 确认类型同步/异步、复制因子等配置参数的影响。经过测试在producer端单个partition的吞吐量通常是在10MB/s左右。在consumer端单个partition的吞吐量依赖于consumer端每个消息的应用逻辑处理速度。因此我们需要对consumer端的吞吐量进行测量。
23、Kafka 分区数可以增加或减少吗,为什么
kafka支持分区数增加
例如我们可以使用 bin/kafka-topics.sh -alter --topic --topic topic-name --partitions 3 命令将原本分区数为1得topic-name设置为3。当主题中的消息包含有key时(即key不为null)根据key来计算分区的行为就会有所影响。当topic-config的分区数为1时不管消息的key为何值消息都会发往这一个分区中当分区数增加到3时那么就会根据消息的key来计算分区号原本发往分区0的消息现在有可能会发往分区1或者分区2中。如此还会影响既定消息的顺序所以在增加分区数时一定要三思而后行。对于基于key计算的主题而言建议在一开始就设置好分区数量避免以后对其进行调整。
Kafka 不支持减少分区数。
按照Kafka现有的代码逻辑而言此功能完全可以实现不过也会使得代码的复杂度急剧增大。实现此功能需要考虑的因素很多比如删除掉的分区中的消息该作何处理如果随着分区一起消失则消息的可靠性得不到保障如果需要保留则又需要考虑如何保留。直接存储到现有分区的尾部消息的时间戳就不会递增如此对于Spark、Flink这类需要消息时间戳(事件时间)的组件将会受到影响如果分散插入到现有的分区中那么在消息量很大的时候内部的数据复制会占用很大的资源而且在复制期间此主题的可用性又如何得到保障与此同时顺序性问题、事务性问题、以及分区和副本的状态机切换问题都是不得不面对的。反观这个功能的收益点却是很低如果真的需要实现此类的功能完全可以重新创建一个分区数较小的主题然后将现有主题中的消息按照既定的逻辑复制过去即可。
24、谈谈你对Kafka生产者幂等性的了解 Kafka精确一次性Exactly-once保障之一 生产者幂等性主要避免生产者数据重复提交至Kafka broker中并落盘。在正常情况下Producer向Broker发送消息Broker将消息追加写到对应的流即某一Topic的某一Partition中并落盘并向Producer返回ACK信号表示确认收到。但是Producer和Broker之间的通信总有可能出现异常如果消息已经写入但ACK在半途丢失了Producer就会进行retry操作再次发送该消息造成重复写入。
为了实现Producer的幂等性Kafka引入了Producer ID即PID和Sequence Number。
PID。每个新的Producer在初始化的时候会被分配一个唯一的PID这个PID对用户是不可见的。Sequence Numbler。对于每个PID该Producer发送数据的每个都对应一个从0开始单调递增的Sequence NumberBroker端在缓存中保存了这seq number,对于接收的每条消息,如果其序号比Broker缓存中序号大于1则接受它,否则将其丢弃,这样就可以实现了消息重复提交了.但是只能保证单个Producer对于同一个的Exactly Once语义 Producer使用幂等性的示例非常简单,与正常情况下Producer使用相比变化不大,只需要 把Producer的配置enable.idempotence设置为true即可,如下所示:
Properties props new Properties();
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
//当enable.idempotence为true时acks默认为 all
// props.put(acks, all);
props.put(bootstrap.servers, localhost:9092);
props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer);
props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer);
KafkaProducer producer new KafkaProducer(props);
producer.send(new ProducerRecord(topic, test);
Prodcuer 幂等性对外保留的接口非常简单其底层的实现对上层应用做了很好的封装应用层并不需要去关心具体的实现细节对用户非常友好
Kafka的幂等性实现了对于单个Producer会话、单个TopicPartition级别的不重不漏也就是最细粒度的保证。如果Producer重启PID发生变化或者写入是跨Topic、跨Partition的单纯的幂等性就会失效需要更高级别的事务性来解决了。当然事务性的原理更加复杂
25、谈谈你对 Kafka事务的了解
幂等性可以保证单个Producer会话、单个TopicPartition、单个会话session的不重不漏如果Producer重启或者是写入跨Topic、跨Partition的消息幂等性无法保证。此时需要用到Kafka事务。Kafka 的事务处理主要是允许应用可以把消费和生产的 batch 处理涉及多个 Partition在一个原子单元内完成操作要么全部完成、要么全部失败。为了实现这种机制我们需要应用能提供一个唯一 id即使故障恢复后也不会改变这个 id 就是 TransactionnalId也叫 txn.idtxn.id 可以跟内部的 PID 1:1 分配它们不同的是 txn.id 是用户提供的而 PID 是 Producer 内部自动生成的并且故障恢复后这个 PID 会变化有了 txn.id 这个机制就可以实现多 partition、跨会话的 EOS 语义。当用户使用 Kafka 的事务性时Kafka 可以做到的保证
跨会话的幂等性写入即使中间故障恢复后依然可以保持幂等性跨会话的事务恢复如果一个应用实例挂了启动的下一个实例依然可以保证上一个事务完成commit 或者 abort跨多个 Topic-Partition 的幂等性写入Kafka 可以保证跨多个 Topic-Partition 的数据要么全部写入成功要么全部失败不会出现中间状态。
事务性示例
Kafka 事务性的使用方法也非常简单用户只需要在 Producer 的配置中配置 transactional.id通过 initTransactions() 初始化事务状态信息再通过 beginTransaction() 标识一个事务的开始然后通过 commitTransaction() 或 abortTransaction() 对事务进行 commit 或 abort示例如下所示生产者
Properties props new Properties();
props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer);
props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer);
props.put(client.id, ProducerTranscationnalExample);
props.put(bootstrap.servers, localhost:9092);
props.put(transactional.id, test-transactional);
props.put(acks, all);
KafkaProducer producer new KafkaProducer(props);
producer.initTransactions();
try {String msg matt test;producer.beginTransaction();producer.send(new ProducerRecord(topic, 0, msg.toString()));producer.send(new ProducerRecord(topic, 1, msg.toString()));producer.send(new ProducerRecord(topic, 2, msg.toString()));producer.commitTransaction();
} catch (ProducerFencedException e1) {e1.printStackTrace();producer.close();
} catch (KafkaException e2) {e2.printStackTrace();producer.abortTransaction();
}
producer.close();
消费者消费者应该设置提交事务的隔离级别
properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,read_committed); Kafka中只有两种事务隔离级别readcommitted、readuncommitted 设置为readcommitted时候是生产者事务已提交的数据才能读取到。在执行 commitTransaction() 或 abortTransaction() 方法前设置为“readcommitted”的消费端应用是消费不到这些消息的不过在 KafkaConsumer 内部会缓存这些消息直到生产者执行 commitTransaction() 方法之后它才能将这些消息推送给消费端应用。同时KafkaConsumer会根据分区对数据进行整合推送时按照分区顺序进行推送。而不是按照数据发送顺序。反之如果生产者执行了 abortTransaction() 方法那么 KafkaConsumer 会将这些缓存的消息丢弃而不推送给消费端应用。设置为read_uncommitted时候可以读取到未提交的数据(报错终止前的数据) 26、Kafka消息是采用Pull模式还是Push模式
push模式下消费者速率主要由生产者决定当消息生产速率远大于消费速率消费者容易崩溃如果为了避免consumer崩溃而采用较低的推送速率将可能导致一次只推送较少的消息而造成浪费。Pull模式可以根据自己的消费能力拉取数据。Push模式必须在不知道下游consumer消费能力和消费策略的情况下决定是立即推送每条消息还是缓存之后批量推送。Pull有个缺点是如果broker没有可供消费的消息将导致consumer不断轮询。但是可以在消费者设置轮询间隔。
27、Kafka缺点
由于是批量发送数据并非真正的实时对于mqtt协议不支持不支持物联网传感数据直接接入仅支持统一分区内消息有序无法实现全局消息有序监控不完善需要安装插件依赖zookeeper进行元数据管理3.0版本去除
28、Kafka什么时候会丢数据
broker端消费丢失
broker端的消息不丢失其实就是用partition副本机制来保证。
unclean.leader.election为true且选举出的首领分区为OSR时 可能就会发生消息丢失min.insync.replicas为N则至少要存在N个同步副本才能向分区写入数据。如果同步副本数量小于N时broker就会停止接收所有生产者的消息、生产者会出现异常如果无法正确处理异常则消息丢失。此时消费者仍然可以读取已有数据、变成只读状态。如果Topic只有一个同步副本那么在这个副本变为不可用时,数据就可能会丢失。kafka的数据一开始是存储在PageCache并定期flush到磁盘上的如果出现断电或者机器故障等PageCache上的数据就丢失了。
生产者端
ack有3种状态保证消息被安全生产 ack0消息传输到Broker端没收到Broker的反馈即发送下一条网络故障导致小东西丢失。ack1如果刚好leader partition挂了数据就会丢失。ackallmin.insync.replicas如果小于N或者Topic只有一个同步副本。消息重试机制未开启。当前消息过大超过max.request.size大小默认为1MB生产者速率超过消费者缓存池空间占满后生产线程阻塞超过最大时间此时生产者会抛出异常如果没有处理好则会丢失数据。
消费者端
enable.auto.committrue消费在处理之前提交了offset则处理异常可能会造成消息的丢失。enable.auto.commitfalseConsumer手动批量提交位点在批量位点中某个位点数据异常时没有正确处理异常而是将批量位点的最后一个位点提交导致异常数据丢失
29、Kafka分区数越多越好吗
并非分区数量越多效率越高
Topic 每个 partition 在 Kafka 路径下都有一个自己的目录该目录下有两个主要的文件base_offset.log 和 base_offset.index。Kafka 服务端的 ReplicaManager 会为每个 Broker 节点保存每个分区的这两个文件的文件句柄。所以如果分区过多ReplicaManager 需要保持打开状态的文件句柄数也就会很多。每个 Producer, Consumer 进程都会为分区缓存消息如果分区过多缓存的消息越多占用的内存就越大n 个分区有 1 个 Leader(n-1) 个 Follower如果运行过程中 Leader 挂了则会从剩余 (n-1) 个 Followers 中选举新 Leader如果有成千上万个分区那么需要很长时间的选举消耗较大的性能。
30、Kafka如何保证消息的有序性
单分区
Kafka在特定条件下可以保障单分区消息的有序性
kafka在发送消息过程中正常情况下是有序的如果消息出现重试则会造成消息乱序。导致乱序的原因是max.in.flight.requests.per.connection默认值为5。 该参数指定了生产者在收到服务器响应之前请求队列中可以提交多少个请求用于提高网络吞吐量。 图中batch1-5在请求队列中batch1作为最新数据进行提交提交失败后如果开启重试机制则batch1会重新添加到本地缓冲池的头部然后提交至请求队列中重新发送。此时batch1的顺序会排在batch5之后发生了乱序。 解决方式是将max.in.flight.requests.per.connection设置为1消息队列中只允许有一个请求这样消息失败后可以第一时间发送不会产生乱序但是会降低网络吞吐量。
或者开启生产者幂等性设置开启后该Producer发送的消息都对应一个单调增的Sequence Number。同样的Broker端也会为每个生产者的每条消息维护一个序号并且每commit一条数据时就会将其序号递增。对于接收到的数据如果其序号比Borker维护的序号大一即表示是下一条数据Broker会接收它否则将其丢弃。如果消息序号比Broker维护的序号差值比一大说明中间有数据尚未写入即乱序此时Broker拒绝该消息Producer抛出InvalidSequenceNumber 如果消息序号小于等于Broker维护的序号说明该消息已被保存即为重复消息Broker直接丢弃该消息Producer抛出DuplicateSequenceNumber Sender发送失败后会重试这样可以保证每个消息都被发送到broker
多分区
Kafka本身无法保障多分区的有序性可以通过业务设计进行保证例如需要单表数据通过自定义partition的方式发送至同一个分区
31、Kafka精确一次性Exactly-once如何保证
宏观上可靠性 at least once 幂等性
具体实现Kafka不丢消息-生产者幂等性-消费者幂等性
详见目录
12、Kafka可靠性保证不丢消息
18、如何保证消息不被重复消费消费者幂等性
24、谈谈你对Kafka生产者幂等性的了解
