【译】99th Percentile Latency at Scale with Apache Kafka

前言

今天带来一篇不是自己原创但觉得很有深度的文章：99th Percentile Latency at Scale with Apache Kafka

---

背景

在同时要求低延时和高质量数据交付的众多Apache Kafka使用场景中，反欺诈、支付系统和股票交易平台仅仅是其中有代表性的几个。例如，网上银行交易就要求反欺诈系统能够实时进行检测且每单交易检测时间不能超过50~100ms，否则会影响用户体验。

对于Kafka而言，数据交付时间（data delivery time）被定义为端到端的延时（end-to-end latency），即消费者获取到生产者发送消息的时间。我们设定延时目标的方式就是确定一个延时值然后满足这个它。比方说，你的延时目标可能是这样表述的：“我希望99%的情况下端到端延时是50ms左右。”

这其实在可用性、持久性和吞吐量三个方面都对你提出了要求。达成高持久性和高吞吐量与实现低延时需要你做一个权衡。主要的挑战是在保持低延时的同时扩展应用程序从而实现高吞吐量。由于延时依赖于你的硬件条件或所选的云厂商，你要有能力监控并调优你的客户端程序以实现特定的延时目标。

之前我们写过一篇文章详细介绍了调优Kafka集群的方法。那篇文章将有助于你直观地理解端到端延时以及在确保延时目标达成的条件下如何配置和扩展你的应用程序。

理解Kafka端到端延时

端到端延时是指应用逻辑调用KafkaProducer.send()生产消息到该消息被应用逻辑通过KafkaConsumer.poll()消费到之间的时间。下图展示了一条消息在此过程中流转的完整路径，包括从内部的Kafka producer到broker，经过备份之后被consumer获取到。

对于端到端延时的构成， 我们定义了5个主要的组成部分：

1. Produce time：内部Kafka producer处理消息并将消息打包的时间
2. Publish time：producer发送到broker并写入到leader副本log的时间
3. Commit time：follower副本备份消息的时间
4. Catch-up time：消费者追赶消费进度，消费到该消息位移值前所花费的时间
5. Fetch time：从broker读取该消息的时间

下面我们分别解释下这5部分各自的含义。特定的客户端配置或应用逻辑设计通常会极大地影响端到端延时，因此我们有必要精准定位哪个因素对延时的影响最大。

Produce time

Produce time衡量的是应用逻辑调用KafkaProducer.send()生产消息到包含该消息的PRODUCE请求被发往leader副本所在broker之间的时间。Kafka producer会将相同topic分区下的一组消息打包在一起形成一个批次（batch）以提升网络I/O性能。默认情况下，producer会立即发送batch，这样一个batch中通常不会包含太多的消息。为了提升打包效果，producer通过设置linger.ms参数人为地暂停一小段时间以等待更多的消息到达并加入到batch中。一旦超过了linger.ms等待时间或batch被填满（batch大小超过了batch.size值），那么batch就被视为“已完成”状态（completed）。

倘若启用了压缩（设置了compression.type），producer会将已完成batch进行压缩。在batch完成之前，batch的大小是基于压缩类型和之前压缩比率估算出来的。

PRODUCE请求被发送到leader副本所在的broker处理之后，broker需要发送应答信息给producer。如果未应答的PRODUCE请求数量超过了阈值（即max.inflight.requests.per.connect，默认是5），那么在producer端进行中的batch就必须等待更长的时间。显然，broker应答的速度越快，producer额外等待的时间就越短。

Publish time

Publish time衡量的是内部producer发送PRODUCE请求到broker与对应消息被写入到leader副本日志之间的间隔。当PRODUCE请求达到broker端时，负责对应连接的网络线程（也就是Processor线程）获取到该请求并将其放置到请求队列上。后面的某个请求处理线程（也就是IO线程或KafkaRequestHandlerThread）从请求队列中取出该PRODUCE请求并处理之。故publish time包含了PRODUCE请求的网络传输时间、在broker上的队列等候时间以及写入到消息日志的时间（准确说是页缓存访问时间）。如果broker端负载不大，网络传输和日志写入时间将会是主要的影响因素。随着broker端负载越来越重，队列等候时间可能会逐渐成为瓶颈。

Commit time

Commit time衡量的是消息被备份到所有ISR副本的时间。Kafka只会将已提交（committed）的消息暴露给consumer，也就是被所有ISR备份过的消息。多个Follower副本从leader副本并行地拉取消息。正常情况下一个健康集群下所有follower副本都应该在ISR集合中。这就意味着消息被commit的时间等于它被备份到ISR中最慢follower副本的时间。

为了备份数据，follower向leader发送FETCH请求——consumer获取消息也是发送FETCH请求。对于副本发出的FETCH请求社区优化了Broker端的默认配置：leader会竭尽所能地提早发送response（只有有1个字节准备好了可以发送，leader就会马上发出去。具体数值由replica.fetch.min.bytes参数或replica.fetch.wait.max.ms参数控制）。Commit time主要受replication factor配置和集群负载影响。

Catch-up time

Kafka中消息是按照其生产的顺序被消费的，除非consumer端显式地变更了消费顺序（通过seek等方法）。同一个分区下，consumer必须要消费完之前发布的消息后才能读取后面的消息。假设消息被提交时，consumer仅仅才消费到该消息的前N条消息。那么Catch-up time就是consumer消费这N条消息的总时间，如下图展示的那样：

当你使用Kafka构建实时应用时，最好让catch-up time越小越好，甚至是0，即消息一旦被提交，consumer就会立刻消费它。如果consumer持续性地落后太多，端到端延时将变得非常不可控。因此，catch-up time的大小依赖于consumer匹配producer吞吐量的能力。

Fetch time

Consumer持续性地调用poll方法来从leader broker获取消息数据。Fetch time衡量的是它从leader副本所在broker获取消息的时间。Broker可能会等待足够多的数据积累到response之后才会返回。默认的配置下，fetch.min.bytes = 0，已经针对延时进行了优化，即只要有数据能够返回立即返回。

端到端延时 VS. producer和consumer延时

下图展示了以Kafka clients端视角观测到的延时，通常被称为producer延时和consumer延时。Producer延时是指调用KafkaProducer.send()与接收到消息已提交应答之间的时间。消息被应答的方式基于acks参数的设置，该参数控制了消息的持久化水平：

• acks = 0：立即返回应答，无需等待broker端的回应
• acks = 1：等待leader副本所在broker写入消息后返回应答
• acks = all：等待所有ISR副本写入消息后返回应答

所以，producer延时由produce time、publish time（acks≠0）、commit time和broker发送response到producer网络传输时间4部分组成。

上图清晰地展示了acks设置对producer延时的影响：通过设置不同的acks，我们可以选择性将producer延时中的某些组成部分移除掉，从而减少producer延时。值得注意的是，无论acks怎么设置，publish time和commit time总是端到端延时的一部分。

Consumer延时是consumer发送FETCH请求到broker与broker返回结果给consumer之间的时间间隔。计算方法是统计程序调用KafkaConsumer.poll()时点和消息返回时点的差值。Consumer延时包含了端到端延时中的Fetch time。

控制端到端延时

如果我们思考一条消息的生命周期，控制端到端延时其实就是控制消息在系统中流转的时间总和。很多Kafka clients端和broker端参数的默认值已然对延时做了优化：比如减少人为等待时间（linger.ms=0）等。其他的延时可能来自于broker端上的队列等候时间，控制这种延时就要设计控制broker的负载（CPU或吞吐量）。

如果我们把系统看做是一个整体，限制端到端延时也需要系统的每一层（producer、broker和consumer）都要能稳定地维持应用逻辑所需的吞吐量要求。举个例子，如果你的应用逻辑按照100MB/秒的速度发送消息，但是由于某种原因，Kafka consumer吞吐量在接下来的几秒时间内下降到10MB/秒，这样绝大多数刚刚被生产出来的消息都需要等待更长的时间才能被消费到。此时，你需要一种高效的方式来扩展你的Kafka clients程序以提升吞吐量——高效地利用broker端资源来减少队列等候时间和偶发的网络拥塞。

限制延时最理想的方式是确保所有延时指标均低于目标值。实际情况中，这种硬性保证几乎很难实现，因为总有各种各样非预期的故障和突发流量。但是，你还是可以小心地设计你的应用程序并调优集群来实现95~99百分位下的延时目标，即控制所有消息95~99%情况下的延时水平。高百分位延时被称为尾延迟（tail latency），因为它们位于延迟段（latency spectrum）的尾部。

目标延时所用的百分位越大，你需要降低或容忍应用最差表现所做的努力就越多。举个例子，偶发的大请求可能会阻塞其后所有的请求，从而抬高后续请求的延时，这就是所谓的队首阻塞（head of line blocking）问题。再比如，大量平素负载请求很低的clients端程序可能会同时向Kafka发送PRODUCE/FETCH请求或同时请求刷新元数据信息，导致broker端请求队列突然拉长，从而引发比平时严重得多的尾延时。这就是所谓的micro-bursting（译者：直接使用原文了。感觉没有特别合适的翻译。微冲击？局部瞬时冲击？）

下面我们分别讨论一下。

评估不同Clients端配置对延时影响的测试环境

我们使用实验结果说明clients参数和吞吐量扩展技术对性能的影响。我们使用Kafka自带的Trogdor测试框架及producer和consumer基准值（ProduceBench&ConsumeBench）进行测试。

所有测试均跑在一个由9台broker构成的Kafka集群上，replication factor设置为3，即在不超过3台broker挂掉的情况下保证没有消息丢失。Kafka broker机器运行在AWS r5.xlarge实例上，EBS为2TB。Broker分布在相同region下的3个不同AZ上以提升容错性。同时，每个topic分区的副本都被放置在不同的AZ上。Kafka clients配置SASL认证和SSL信道加密，但broker之间的通讯依然使用PLAINTEXT。

最后，除非显式提及，否则我们所有的测试都使用以下的clients端配置：

Replication factor	3
topic分区数	108
security.protocol	SASL_SSL
acks	all
linger.ms	5
compression.type	LZ4
消息大小	512B
消息Key大小	4B
Trogdor消息值生成器	uniformRandom
Trogdor消息Key生成器	sequential
Trogdor实例数	9

这个测试场景确实会引人额外的延时：多个AZ的部署架构会增加commit time，因为引入了跨AZ的备份。另外无论是clients端还是broker端，SSL加密也是有开销的。最后由于SSL无法利用Zero Copy特性进行数据传输，因为consumer获取消息时也会增加额外的开销。虽然这些都会影响延时，但这毕竟符合真实的使用场景，因此我们决定采用这样的部署结构进行测试。

持久性设置对延时的影响

当我们拿延时目标去PK其他指标时，我们最好先考虑持久性。我们通常都要保证满足一定程度的持久性，毕竟你的数据是非常关键的。优化持久性会增加端到端延时，因为它增加了备份的代价（commit time），并且给broker添加了备份方面的负载 ，而这会增加队列等候时间。

Replication factor

Replication factor（rf）是Kafka持久化保证的核心，它定义了Kafka集群上保存的topic副本数。Replication factor = rf表示我们最多能够容忍rf - 1台broker宕机而不必数据丢失。Rf = 1能够令端到端延时最小化，但却是最低的持久化保证。

增加rf会增加备份开销并给broker额外增加负载。如果clients端带宽在broker端均匀分布，那么每个broker都会使用rf * w写带宽和r + (rf - 1) * w读带宽，其中w是clients端在broker上的写入带宽占用，r是读带宽占用。由此，降低rf对端到端延时影响的最佳方法就是确保每个broker上的负载是均匀的。这会降低commit time，因为commit time是由最慢的那个follower副本决定的。如果我们能减少各个broker延时之间方差，我们就能降低整体的尾延时水平。

如果你的Kafka broker使用了过多的磁盘带宽或CPU，follower就会开始出现追不上leader的情况从而推高了commit time。我们建议为副本同步消息流量设置成使用不同的listener来减少与正常clients流量的干扰。你也可以在follower broker上增加I/O并行度，并增加副本拉取线程数量（number.replica.fetchers）来改善备份性能。

acks

纵然我们配置了多个副本，producer还是必须通过acks参数来配置可靠性水平。设置acks=all能够提供最强的可靠性保证，但同时也会增加broker应答PRODUCE请求的时间，就像我们之前讨论的那样。

Broker端应答的速度变慢通常会降低单个producer的吞吐量，进而增加producer的等待时间。这是因为producer端会限制未应答请求的数量（max.inflight.requests.per.connection）。举个例子，在我们的环境中acks=1，我们启动了9个producer（同时也跑了9个consumer），吞吐量达到了195MB/秒。当acks切换成all时，吞吐量下降到161MB/秒。设置更高级别的acks通常要求我们扩展producer程序才能维持之前的吞吐量水平以及最小化producer内部的等待时间。

min.insync.replicas

min.insync.replicas是一个重要的持久化参数，因为它定义了broker端ISR副本中最少要有多少个副本写入消息才算PRODUCE请求成功。这个参数会影响可用性，但是不会影响端到端的延时。无论min.insync.replicas取何值，消息都必须要被复制到ISR所有副本中。因此，选择一个小一点的值并不能减少commit time或是延时。

在满足延时目标的前提下扩展吞吐量

延时-吞吐量权衡

优化Kafka clients端吞吐量意味着优化打包的效果（batching）。Kafka producer内部会执行一类打包，即收集多条消息到一个batch中。每个batch被统一压缩然后作为一个整体被写入日志或从日志中读取。这说明消息备份也是以batch为单位进行的。打包化会减少每条消息的成本，因为它将这些成本摊还到clients端和broker端。通常来说，batch越大这种开销降低的效果就越高，减少的网络和磁盘I/O就越多。

另一类打包就是在单个网络请求/响应中收集（打包）多个batch以减少网络数据传输量。这能降低clients端和broker端的请求处理开销。

这类打包能够提升吞吐量和降低延时，因为batch越大，网络传输I/O量越小，CPU和磁盘使用率越低，故最终能够优化吞吐量。另外batch越大还能减低端到端延时，因为每条消息的成本降低了，使得系统处理相同数量消息的总时间变少了。

这里的延时-吞吐量权衡是指通过人为增加等待时间来提升打包消息的能力。但过了某个程度，人为等待时间的增加可能会抵消或覆盖你从打包机制获得的延时收益。因此你的延时目标有可能会限制你能实施打包化的水平，进而减少所能达到的吞吐量并增加延时。如果拉低了本能达到的吞吐量或端到端延时水平，你可以通过扩展集群来换取或“购买”更多的吞吐量或处理能力。

配置Kafka producer和consumer实现打包化

对于producer而言，打包化水平由两个参数控制：batch.size（默认值16KB）——限制每个batch的大小；linger.ms（默认值0）——限制人为等候时间。如果你的应用程序发送消息的速度很高，那么即使linger.ms=0，batch也能在被发送之前快速填满。反之，如果发送速率很低，那么你就需要增加linger.ms值以提升打包水平。

对于consumer而言，你能够调整每个FETCH响应中返回的数据量。具体方法是调优fetch.min.bytes（默认值1）——该参数指定了broker端能够发送FETCH响应所能包含的最小数据字节数。另外还有一个参数fetch.max.wait.ms（默认值500ms）用于控制broker端等待数据的超时时间。FETCH响应中包含更多的数据会使得后续的FETCH请求数变少。Producer端打包化间接影响PRODUCE和FETCH请求的数量，因为batch定义了数据能够被获取的最小数据量。

值得注意的是，默认情况下，Kafka producer和consumer设置的是无人为等待时间，这么做的目的是为了降低延时。但是，即使你的目标就是了使延时最小化，我们依然推荐你设置一个不为0的linger.ms值，比如5~10ms。当然，这么做是有前提的：

• 如果你扩展了你的producer程序，平均下来使得每个producer实例的发送速率变得很低，那么你的batch只会包含很少的几条消息。如果你整体的吞吐量已然很高了，那么你可能会直接把你的Kafka集群压挂，导致超高的队列等候时间从而推高延时。此时，设置一个较小的linger.ms值确实能够改善延时。
• 如果你在意尾延时，那么增加linger.ms可能会降低请求速率以及同时到达broker端的瞬时冲击流量。这种冲击越大，请求在尾部的延时就越高。这些瞬时冲击流量决定了你的尾延时水平。因此增加linger.ms能够显著降低尾延时，即使你平均的延时会有所增加，比如平均增加了linger.ms等待时间。

下面这个实验说明了以上两种场景。我们启动了90个producer，向一个有108个分区的topic发送消息。生产端整体的吞吐量峰值在90MB/秒。我们跑了3次测试，每一次对应一种不同的producer配置。下图展示了50百分位和99百分位水平下的producer延时。由于我们设置了acks=all，producer延时包含produce time、publish time和commit time。

鉴于我们使用了超多的producer实例来计算整体吞吐量，linger.ms = 0的设置基本上会驱散producer端的打包化效果。把linger.ms值提升到5ms会极大地改善打包化水平。PRODUCE请求数量从2800骤降到1100。这种变化会同时降低两个百分位水平下的producer延时，而99百分位下的降低效果更为显著。

增加batch.size并未直接影响producer端的等待时间，因为producer在积累batch的等待时间不会超过linger.ms值。在我们的实验中，增加batch.size到128KB并未提升打包化，因为单个producer的吞吐量其实是很低的。就像我们期望的那样，producer延时在两次实验中没有太多改进。

总结一下，如果你的目标是为了降低延时，我们推荐保持默认的batch参数并适当增加linger.ms，前提是producer无法进一步打包更多的消息。如果你在意尾延时，最好调优下打包水平来减少请求发送率以及大请求冲击的概率。

不增加人为等待下提升打包效果 

打包效果不好的另一个原因是producer发送消息给大量分区。如果消息不是发往同一个分区的，它们就无法聚集在一个batch下。因此，通常最好设计成让每个producer都只向有限的几个分区发送消息。

另外，可以考虑升级到Kafka 2.4 producer。这个版本引入了一个全新的Sticky分区器。该分区器能够改善non-keyed topic的打包效果，同时还无需引入人为等待。

Clients数量对尾延时的影响

即使整体的生产和消费的吞吐量保持不变，通常也是Clients数越多，broker上负载越大。这是因为clients数量多会导致更多的METADATA请求发到Kafka，继而要维护更多的连接，故给broker带来更大的开销。

相对于50百分位或平均延时的影响，Clients数量增加对尾延时的影响更大。每个producer最多发送max.inflight.requests.per.connection个PRODUCE请求给单个broker，而每个consumer一次最多只会给一个broker发送FETCH请求。Clients越多，同一时刻发送到broker的PRODUCE和FETCH请求也就越多，这就增加了形成请求瞬时冲击的概率，进而推高了尾延时。

Consumer数量通常由topic分区数量以及期望consumer没有较大lag的目标共同决定。但是，我们却很容易为了扩展吞吐量而引入大量的producer。基于吞吐量的考量增加producer实例数可能有相反的效果，因为producer会导致更少的消息被打包，毕竟每个producer处理了更少的消息，因而发送速率会变慢。同时producer还必须等待更长的时间来积累相同数量的消息进到batch里面。

在我们的实验中，我们将producer的数量从90增加到900，发现吞吐量没有他打变化：90MB/秒。我们使用batch.size=16KB，linger.ms = 5，acks=all的设置。下图展示了三次实验的producer延时：

结果显示增加producer数量（90 =》 900）增加了60%的中位数延时值，而99百分位延时值几乎增加了3倍！延时的增加是因为producer端打包效果变差导致的。尾延时的增加是因为更大的请求瞬时冲击，这会拉升broker端延时，同时producer端会等待更长的时间来接收应答。在900个producer的测试中，broker完全被PRODUCE请求压垮了。用于处理请求的时间几乎占到了broker端CPU使用率的100%。另外由于我们使用了SSL，它也会进一步引入请求级的开销。

如果你通过添加producer来提升吞吐量，那么可以考虑增加单个proudcer的吞吐量——方法是改善打包效果。不管怎样，你最终可能会有很多producer实例。比如，大公司收集设备上的统计指标，而设备数可能有成千上万。此时，你可以考虑使用一个代理来收集来自多个clieints的请求，然后把它们转换成更高效的PRODUCE请求再发给Kafka。你也可以增加broker数来降低单个broker上的请求负载。

扩展consumer

当扩展consumer时，记住：同一个group下的所有consumer实例发送位移提交和心跳请求给同一个broker，也就是Coordinator所在的broker。Consumer数越多，位移提交间隔（auto.commit.interval.ms）内需要提交位移的频次也就越多。位移提交本质上就是发送给__consumer_offsets主题的PRODUCE请求。因此增加consumer数量会导致broker上的请求负载增加，特别是auto.commit.interval.ms值很小的时候。

压缩设置的影响

默认情况下，Kafka producer不做压缩。compression.type参数决定要不要做压缩。压缩会在producer端引入额外的开销来压缩消息，在broker端做校验时解压缩从而引入额外的开销，另外在consumer端解压缩也是开销。Broker端的压缩参数compression.type应该设成“producer”以避免重新压缩的成本——这样，broker直接将原始已压缩消息写入日志即可。

虽然压缩会增加CPU开销，但它还是可能减少端到端延时的，因为它能显著地降低处理数据所需的带宽占用，进而减少broker端的负载。压缩是在batch级别上完成的，故打包效果越好，压缩效果也就越好。

更多的分区可能增加延时

topic分区是调节Kafka并行度的最小单位。不同分区的消息能够被producer并行发送，并行写入到不同的broker上，同时也可以被不同的consumer并行读取。因此分区越多吞吐量越大。单纯从吞吐量的角度来看，每个broker上10个分区就能够达到最大限度的吞吐量了。你可能需要更多的分区数来支撑你的应用逻辑。

太多的分区数对端到端延时有负面影响。每个topic分区数太多通常会导致producer端更差的打包效果。每个broker上分区太多会使得follower副本产生更多的FETCH请求。每个FETCH请求必须要遍历所有要同步的分区，而leader必须要检查FETCH请求中每个分区的状态和可返回的数据。这些操作都是很小的磁盘IO。因此太多的分区会导致更大的commit time以及更高的CPU负载，进而引起更长的队列等候时间。

Commit time的增加和CPU负载变高会推高所有clients的端到端的延时，即使是那些只与一小部分分区交互的clients。

我们做了这样的一个测试：两个topic。一个topic有9个producer，每秒发送5MB数据，同时被由9个consumer组成的group消费。该实验运行了几天，之后我们增加了分区数，逐步将其从108提升到了7200（每台broker上800个分区），每一步都运行一个小时。第二个topic在整个实验阶段保持9个分区。同时启动9个producer，每一个产生消息到一个对应的分区。另外还有一个9consumer实例构成的group。这些producer的发送速率是512B/秒。

下图展示了分数区对99百分位端到端延时的影响。随着每个broker上分区数的增加，clients的端到端延时大致呈线性增加趋势。分区数的增加会推高broker上的CPU负载同时拖慢所有clients的备份，即使是对那些只与固定分区数量交互的clients而言，也会抬高端到端延迟。

 为了减少延时，最好还是限制每个broker上的分区数，方法是减少总的分区数或扩展集群。你还可以通过增加fetcher线程数量的方式来改善commit time。

Broker端负载对延时的影响

我们之前讨论了broker上负载会增加队列等候时间，从而推高端到端延时。很容易看到请求速率的增加也会使队列等候时间拉长，因为请求数越多，队列越长。

Broker端高资源利用率（如disk、CPU）都会增加队列等候时间，而这种等候时间一般是指数级增长。这就是队列理论中提到的一个著名特性：Kingman公式证明等待某种资源的时间正比于(资源繁忙时间百分比/资源空闲时间百分比）

由于延时随着资源利用率指数级增长，当broker端某种资源接近100%使用率时你会观测到超高的延时出现。减少资源使用率的方法是减少资源使用（比如减少连接数，请求数和分区数）或者是扩展集群，这些方法都能有效地降低延时。同时，让负载均匀地在各个broker上进行分布也是有帮助的。这对于我们改善尾延时非常有效。

如果你使用的是Confluent Cloud，你不用担心如何维护broker的事情——我们已经帮你做了。我们会自动地检测broker端上的资源使用率情况，然后自动为你执行负载均衡操作以维持各个broker上的负载均匀。 

总结

本文论证了如果要扩展clients和分区来提升吞吐量同时又要限制延时不要恶化，我们必须要限制每个broker上的资源使用率——比如限制连接数、分区数和请求速率等。如果要优化尾延时则要求我们将任何来自clients端的瞬时冲击降到最低。均匀分布负载对于最小化尾延时同样重要，因为尾延时是由最慢的broker决定的。