为何要划分优先级
Kafka处理请求不区分优先级,但这种绝对公平的策略有时会发生问题。 比如:创建一个单分区双副本的主题,当时集群中的Broker A机器保存了分区的Leader副本,Broker B保存了Follower副本。突然业务激增,Broker A瞬间积压大量未处理PRODUCE请求。运维“不凑巧”执行了次Preferred Leader选举,将Broker B显式地调整成了Leader。
若Producer把acks设为all,那在LeaderAndIsr请求(负责调整副本角色,比如Follower和Leader角色转换)之前积压的那些PRODUCE请求就无法正常完成,因为这些请求要一直等待ISR中所有Follower副本同步完成。 但此时,Broker B成为了Leader,它上面的副本停止了拉取消息,这就可能出现一种结果:这些未完成的PRODUCE请求会一直保存在Broker A上的Purgatory缓存中。Leader/Follower的角色转换,导致无法完成副本间同步,所以这些请求无法被成功处理,最终Broker A抛出超时异常,返回给Producer程序。
Purgatory缓存是Broker端暂存延时请求的地方。
这就是因为对请求不区分优先级。
1 基本概念
1.1 Data plane & Control plane
Data plane和Control plane的字面意思是数据面和控制面,各自对应数据类请求和控制类请求。 Kafka请求类型分为:
- 数据类请求 Data plane负责处理数据类请求
- 控制类请求 Control plane负责处理控制类请求
Controller与Broker交互的请求类型有如下种类:
- LeaderAndIsrRequest
- StopReplicaRequest
- UpdateMetadataRequest
都属控制类请求,通常应该被赋予高优先级,比如PRODUCE和FETCH请求。
SocketServer负责对这两大类请求区分处理。
1.2 监听器(Listener)
区分数据类请求和控制类请求不同处理方式的主要途径。即创建多组监听器分别执行数据类和控制类请求的处理。
- Kafka Broker端如下参数用以配置监听器
监听器定义为
EndPoint类
属性
- host Broker主机名。
- port Broker端口号。
- listenerName
监听器名称。目前预定义的名称包括
- PLAINTEXT
- SSL
- SASL_PLAINTEXT
- SASL_SSL
Kafka允许自定义其他监听器名称,比如CONTROLLER、INTERNAL。
- securityProtocol:监听器使用的安全协议。Kafka支持4种安全协议
- PLAINTEXT
- SSL
- SASL_PLAINTEXT
- SASL_SSL
Broker端参数
比如若Broker端参数配置如下:
配置3个监听器,分别是CONTROLLER、INTERNAL和EXTERNAL,使用的安全协议分别是PLAINTEXT、PLAINTEXT和SSL。
那SocketServer如何实现Data plane与Control plane分离的呢?
SocketServer
属性
- 实现请求优先级相关的字段
对于Data plane,线程池的说法没有问题,因为Processor线程确实有很多,而Acceptor也可能多个,因为SocketServer会为每个EndPoint(每套监听器)创建一个对应的Acceptor线程。
但Control plane不同。 Control plane那组属性变量都是以Opt结尾的,即Option类型,完全可以不使用Control plane,即你可让Kafka不区分请求类型,2.2.0之前设计就是这样。
但一旦开启Control plane设置,其Processor线程和Acceptor线程都是1个。 它对应的RequestChannel里面的请求队列长度被硬编码成20,即控制类请求的数量应该远小于数据类请求,因而不需要为它创建线程池和较深的请求队列。
创建Data plane所需资源
- 负责为Data plane创建所需资源
- 执行流程
- 最大连接数计数器将被用来确保没有配额超限的情形发生
- 创建Processor线程池。对于Data plane而言,线程池的数量由Broker端参数num.network.threads决定
- 将<监听器,Acceptor线程>对加入到Acceptor线程池统一管理
比如配置listeners=PLAINTEXT://localhost:9092, SSL://localhost:9093,默认会为PLAINTEXT和SSL监听器分别创建一个Acceptor线程和一个Processor线程池。
具体为哪些监听器创建依据配置而定,Kafka只会为Data plane所使的监听器创建这些资源。
创建Control plane所需资源
基于控制类请求远小于数据类请求假设,Control plane的配套资源只有1个Acceptor线程 1个Processor线程 1个深度是20的请求队列而已。和Data plane相比,这些配置稍显寒酸,够用就行。 和createDataPlaneAcceptorsAndProcessors类似,只是需判断是否配置用于Control plane的监听器。 目前,Kafka规定只能有1套监听器用于Control plane,而不能像Data plane那样可以配置多套。
- 以上都没听提到启动Acceptor和Processor线程。那这些线程到底是在什么时候启动? 在启动SocketServer组件之后启动的,具体代码在KafkaServer.scala文件的startup方法
Broker端参数control.plane.listener.name用于设置Control plane所用的监听器。
在默认情况下,这个参数的值是空(Null)。Null的意思就是告诉Kafka不要启用请求优先级区分机制,但如果你设置了这个参数,Kafka就会利用它去listeners中寻找对应的监听器了。
假设Broker端配置:
代码语言:javascript复制listener.security.protocol.map=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:SSL
listeners=CONTROLLER://192.1.1.8:9091,INTERNAL://192.1.1.8:9092,EXTERNAL://10.1.1.5:9093
control.plane.listener.name=CONTROLLER
名字是CONTROLLER的那套监听器将被用于Control plane。 名字是INTERNAL和EXTERNAL的这两组监听器用于Data plane。
Kafka如何知道CONTROLLER这套监听器给Control plane使用? KafkaConfig类封装了Broker端所有参数信息和便捷的工具方法。
确认Control plane监听器完整的查找逻辑:
- 先获取Broker端参数
control.plane.listener.name
值。该实例中,值是字符串CONTROLLER - 读取Broker端参数
listener.security.protocol.map
值,并找出CONTROLLER对应的安全认证协议。该例中,CONTROLLER对应安全认证协议PLAINTEX - controlPlaneListenerName方法的作用是拿到这组值,即对
- 最后,controlPlaneListener方法拿到这组值,取出监听器名称CONTROLLER去寻找Broker端参数listeners中对应的监听器。该例中,这个监听器就是
CONTROLLER://192.1.1.8:9091
。
getControlPlaneListenerNameAndSecurityProtocol
拿参数值后,controlPlaneListener会记录这个值,传到SocketServer#createControlPlaneAcceptorAndProcessor。这样,SocketServer就能知道你到底有没有为Control plane设置专属监听器了。
和请求优先级之间的关系
严格说,Kafka没有为请求设置数值型的优先级,因此,我们并不能把所有请求按照所谓的优先级进行排序。 Kafka仅实现了粗粒度的优先级处理,即整体上把请求分为
- 数据类请求
- 控制类请求
而且没有为这两类定义可相互比较的优先级。那如何把刚刚说的所有东西和这里的优先级进行关联呢? 定义了多套监听器 & 底层处理线程的方式来 以区分这俩类请求。虽然无法直接比较俩类请求的优先级,但在实际应用中,由于数据类请求数量>>控制类请求,因此,为控制类请求单独定义处理资源的做法,实际上就等同于拔高了控制类请求的优先处理权。 这也算间接实现了优先级的区别对待。
总结
Data plane:负责处理数据类请求,这类请求通常不需要高优先级处理。 Control plane:负责处理控制类请求,这类请求需要高优先级处理。 监听器:Kafka允许Broker定义多套监听器,每套监听器可用于Data plane或Control plane。 优先级实现原理:你要知道的是,社区设计了两套资源分别处理Data plane和Control plane请求。