Kafka自0.9.0.0版本引入了配额管理(quota management),旨在broker端对clients发送请求进行限流(throttling)。目前Kafka支持两大类配额管理:
- 网络带宽(network bandwidth)配额管理:定义带宽阈值来限制请求发送速率,阈值单位是字节/秒,即bytes/s。该功能是0.9.0.0版本引入的
- CPU配额管理:定义CPU使用率阈值来限制请求发送速率,阈值以百分比的形式给出,如quota = 50表示50%的CPU使用率上限。该功能是0.11.0.0版本引入的
本文主要讨论网络带宽配额管理。关于CPU配额管理的部分我们将在下一篇中进行讨论。
一、配额能做什么?
设置了基于带宽的配额之后,Kafka能够做到:
1. 限制follower副本拉取leader副本消息的速率
2. 限制producer的生产速率
3. 限制consumer的消费速率
二、配额作用域
目前可以在3个层级上设置配额:
1. client id
2. user
3. user + client id
第一种是client id,即新版本clients端的通用参数client.id设置的值。一旦为某个client id设置了配额,所有具有该client id的clients段程序都将受到该配额的限制。
第二种是user,为启用认证之后位于认证体系中的某个用户主体(user principal),比如一个Kerberos用户:user1/kafka.host1.com@REALM,Kafka解析出来的用户名是'user1’。当然我们可以设置sasl.kerberos.principal.to.local.rules参数修改这种解析规则,不过这不在本文的讨论范围内。
第三种就是user + client id,实际上是包含前两种的一个二元组。它是最细粒度化的配额管理作用域。
当然,这3种作用域还可以设置各自的默认值配额(默认是没有配额的,即默认值通常是无穷大),包括:client id作用域默认值、user作用域默认值、user + client id作用域默认值,其中最后一项又可细分为4个子作用域,即
- user作用域默认值 + client id作用域指定值
- user作用域指定值 + client id作用域指定值
- user作用域默认值 + client id作用域默认值
- user作用域指定值 + client id作用域默认值
因此,实际上总共有8种可能的配额作用域设置值,它们的优先级关系依次如下(从高到低):
- user作用域指定值 + client id作用域指定值(即为user + client id设置了特定值配额)
- user作用域指定值 + client id作用域默认值(为user设置了特定值配额,为client id设置了默认值配额)
- user作用域(为user设置了特定值配额)
- user作用域默认值 + client id作用域指定值(为user设置了默认值配额,为client id设置了特定值配额)
- user作用域默认值 + client id作用域默认值(为user和client id设置了默认值配额)
- user作用域默认值(为user设置了默认值配额)
- client id作用域(为client id设置了特定值配额)
- client id作用域默认值(为client id设置了默认值配额)
当多条配额规则冲突时我们可以根据以上规则确定应用的是哪一条。举个例子,我们为user = 'good-user'的用户配置了100MB/s的配额,同时为[user='good-user', client id = 'producer-1']设置配额为50MB/s,那么当good-user用户使用名为‘producer-1’的producer发送消息时Kafka保证它的请求处理速率不会超过50MB/s,即第二条规则覆盖了第一条规则。
三、如何设置?
我们根据第一小节中提到的3种功能来分别讨论。
3.1 限制follower副本拉取leader副本消息的速率
方法1: 设置broker端动态参数leader.replication.throttled.rate和follower.replication.throttled.rate。
前者控制leader副本端处理FETCH请求的速率,后者控制follower副本发送FETCH请求的速率。既然是动态参数,说明它们的值可以实时修改而无需重启broker。假设我要为broker 0和1设置follower和leader限速为100MB/s,方法如下:
执行下列命令检查下是否配置成功:
同时,还必须为topic设置leader.replication.throttled.replicas和follower.replication.throttled.replicas 。这组参数需要指定要限速的副本,如果想让topic的所有副本都生效,可以使用*通配符:
方法2:执行分区重分配时设置
在使用kafka-reassign-partitions.sh(bat)脚本执行分区重分配时也可以设定 ,方法如下(依然设置成100MB/s):
实际上,该脚本通过--throttle参数间接设置了leader.replication.throttled.rate和follower.replication.throttled.rate参数,故本质上和方法1是相同的。值得注意的,该脚本只会对参与到分区重分配的broker设置配额,对其他broker是不起作用的。
3.2 限制producer端速率
方法1: 设置broker端静态参数quota.producer.default参数
比如:在server.properties中加入quota.producer.default=15728640将限制所有连入该broker的producer的TPS全部降到15MB/s以下。设置此参数的好处是能够限制集群上的所有producer,但劣处也在于此,对所有producer“一视同仁”,无法细粒度地对个别clients进行设置。故社区在0.11.0.0版本将其标记为"Deprecated",并始终推荐用户使用动态参数的方式来为producer端进行限速。
方法2:设置动态参数producer_byte_rate
首先演示为所有client id设置默认值,假设我们为所有producer程序设置其TPS不超过20MB/s,即20971520B/s,命令如下:
然后我们为client.id=‘producer-1'的producer单独设置其TPS不超过10MB/s,即10485760B/s,命令如下:
下面做个简单的试验验证下,我们启动两个client.id是producer-1和producer-2的producer程序去验证它们的TPS小于设置的阈值:
可见producer-2的TPS被限制在了20MB/s以下。接下来我们试试producer-1(因为其阈值设置得小,故这次我们少发一些消息以加速整个试验进程):
为user设置配额的方法与client id类似,设置全局默认值:
为特定用户(user1)设置:
最后是设置(user + client id)作用域设置配额。依然是全局默认值:
然后是特定值:
3.3 限制consumer端速率
和producer类似,也存在两种方法:
方法1: 设置broker端静态参数quota.consumer.default参数
与quota.producer.default完全相同的参数,只不过是适用于consumer端的,不再赘述。
方法2:设置动态参数consumer_byte_rate
与producer_byte_rate完全对等的参数,只是适用于consumer端的,不再赘述
四、配额算法
简单来说,我们假设当前实际值是O,T是我们设置的阈值,而W表示某一段时间范围,我们希望在W时间内O能够下降到T以下(如果O本来就比T小,则什么都不用做),那么broker端就需要延缓等待一段时间。如果假设这段时间是X,那么以下等式成立:
O * W = (W + X) * T
由此得出X = (O - T) / T * W。这就是Kafka用于计算等待时间的公式。当然在具体实现时,Kafka提供了两个参数来共同计算W:W = quota.window.num * quota.window.size.seconds。前者表示取样的时间窗口个数,后者表示时间窗口大小。特别是后者会在CPU配额管理中用到。不过在本文中,我们可以统一使用W即可。当Kafka检测到配额透支情况发生时,broker不会返回错误而是直接将超支配额的客户端进行减速处理。它会计算需要X然后令client强制sleep直至令其降到配额之下。该方法对于client来说完全透明。同时,client也不需要自己实现任何特殊的策略来应对。事实上,有的client在应对这种情况时会不停地重试反而加剧了本要解决的问题。
五、可能的问题
如前所述,限速是在broker端执行的!broker端故意“sleep”来限速的做法虽然对clients端透明,但确实也会引起clients端请求的超时,故在实际使用过程中适当地增加request.timeout.ms对于启用了限速的Kafka环境而言就显得非常必要了。