深入浅出系列之 -- kafka调优

左手的ㄟ右手 2021-11-05 13:22 354阅读 0赞

**背景引入：很多同学看不懂kafka参数**

>     今天给大家聊一个很有意思的话题，大家知道很多公司都会基于Kafka作为MQ来开发一些复杂的大型系统。而在使用Kafka的客户端编写代码与服务器交互的时候，是需要对客户端设置很多的参数的。所以我就见过很多同学，可能刚刚加入团队，对Kafka这个技术其实并不是很了解。此时就会导致他们看团队里的一些资深同事写的一些代码，会看不懂是怎么回事，不了解背后的含义，这里面尤其是一些**Kafka参数的设置**。
> 
>     所以这篇文章，我们还是采用老规矩画图的形式，来聊聊Kafka生产端一些常见参数的设置，让大家下次看到一些Kafka客户端设置的参数时，不会再感到发怵。

下面根据一段java代码进行调优：

# **一段Kafka生产端的示例代码** #

Properties props = new Properties();
    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");
    props.put("buffer.memory", 67108864);
    props.put("batch.size", 131072);
    props.put("linger.ms", 100);
    props.put("max.request.size", 10485760);
    props.put("acks", "1");
    props.put("retries", 10);
    props.put("retry.backoff.ms", 500);
    
    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

### **内存缓冲的大小** ###

> 首先我们看看“buffer.memory”这个参数是什么意思？
> 
>     Kafka的客户端发送数据到服务器，一般都是要经过缓冲的，也就是说，你通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集成一个一个的Batch，再发送到Broker上去的。
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70][]
> 
> 所以这个“buffer.memory”的本质就是用来约束KafkaProducer能够使用的内存缓冲的大小的，他的默认值是**32MB**。那么既然了解了这个含义，大家想一下，在生产项目里，这个参数应该怎么来设置呢？你可以先想一下，如果这个内存缓冲设置的过小的话，可能会导致一个什么问题？
> 
> 首先要明确一点，那就是在内存缓冲里大量的消息会缓冲在里面，形成一个一个的Batch，每个Batch里包含多条消息。
> 
> 然后KafkaProducer有一个Sender线程会把多个Batch打包成一个Request发送到Kafka服务器上去。
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 1][]
> 
> 那么如果要是内存设置的太小，可能**导致一个问题**：消息快速的写入内存缓冲里面，但是Sender线程来不及把Request发送到Kafka服务器。
> 
> 这样是不是会造成内存缓冲很快就被写满？一旦被写满，就会阻塞用户线程，不让继续往Kafka写消息了。
> 
> 所以对于“buffer.memory”这个参数应该结合自己的实际情况来进行压测，你需要测算一下在生产环境，你的用户线程会以每秒多少消息的频率来写入内存缓冲。
> 
> 比如说每秒300条消息，那么你就需要压测一下，假设内存缓冲就32MB，每秒写300条消息到内存缓冲，是否会经常把内存缓冲写满？经过这样的压测，你可以调试出来一个合理的内存大小。

### **多少数据打包为一个Batch合适？** ###

> 接着你需要思考第二个问题，就是你的“batch.size”应该如何设置？这个东西是决定了你的每个Batch要存放多少数据就可以发送出去了。
> 
> 比如说你要是给一个Batch设置成是16KB的大小，那么里面凑够16KB的数据就可以发送了。这个参数的默认值是16KB，一般可以尝试把这个参数调节大一些，然后利用自己的生产环境发消息的负载来测试一下。
> 
> 比如说发送消息的频率就是每秒300条，那么如果比如“batch.size”调节到了32KB，或者64KB，是否可以提升发送消息的整体吞吐量。
> 
> 因为理论上来说，提升batch的大小，可以允许更多的数据缓冲在里面，那么一次Request发送出去的数据量就更多了，这样吞吐量可能会有所提升。
> 
> 但是这个东西也不能无限的大，过于大了之后，要是数据老是缓冲在Batch里迟迟不发送出去，那么岂不是你发送消息的延迟就会很高。
> 
> 比如说，一条消息进入了Batch，但是要等待5秒钟Batch才凑满了64KB，才能发送出去。那这条消息的延迟就是5秒钟。
> 
> 所以需要在这里按照生产环境的发消息的速率，调节不同的Batch大小自己测试一下最终出去的吞吐量以及消息的 延迟，设置一个最合理的参数。

**要是一个Batch迟迟无法凑满怎么办？**

> 要是一个Batch迟迟无法凑满，此时就需要引入另外一个参数了，“**linger.ms**”
> 
> 他的含义就是说一个Batch被创建之后，最多过多久，不管这个Batch有没有写满，都必须发送出去了。
> 
>      给大家举个例子，比如说batch.size是16kb，但是现在某个低峰时间段，发送消息很慢。这就导致可能Batch被创建之后，陆陆续续有消息进来，但是迟迟无法凑够16KB，难道此时就一直等着吗？当然不是，假设你现在设置“linger.ms”是50ms，那么只要这个Batch从创建开始到现在已经过了50ms了，哪怕他还没满16KB，也要发送他出去了。所以“linger.ms”决定了你的消息一旦写入一个Batch，最多等待这么多时间，他一定会跟着Batch一起发送出去。避免一个Batch迟迟凑不满，导致消息一直积压在内存里发送不出去的情况。**这是一个很关键的参数。  **
> 
> 这个参数一般要非常慎重的来设置，要配合batch.size一起来设置。
> 
> 举个例子，首先假设你的Batch是32KB，那么你得估算一下，正常情况下，一般多久会凑够一个Batch，比如正常来说可能20ms就会凑够一个Batch。
> 
> 那么你的linger.ms就可以设置为25ms，也就是说，正常来说，大部分的Batch在20ms内都会凑满，但是你的linger.ms可以保证，哪怕遇到低峰时期，20ms凑不满一个Batch，还是会在25ms之后强制Batch发送出去。
> 
> 如果要是你把linger.ms设置的太小了，比如说默认就是0ms，或者你设置个5ms，那可能导致你的Batch虽然设置了32KB，但是经常是还没凑够32KB的数据，5ms之后就直接强制Batch发送出去，这样也不太好其实，会导致你的Batch形同虚设，一直凑不满数据。

### **最大请求大小** ###

> “max.request.size”这个参数决定了每次发送给Kafka服务器请求的最大大小，同时也会限制你一条消息的最大大小也不能超过这个参数设置的值，这个其实可以根据你自己的消息的大小来灵活的调整。
> 
> 给大家举个例子，你们公司发送的消息都是那种大的报文消息，每条消息都是很多的数据，一条消息可能都要20KB。
> 
> 此时你的batch.size是不是就需要调节大一些？比如设置个512KB？然后你的buffer.memory是不是要给的大一些？比如设置个128MB？
> 
> 只有这样，才能让你在大消息的场景下，还能使用Batch打包多条消息的机制。但是此时“max.request.size”是不是也得同步增加？
> 
> 因为可能你的一个请求是很大的，默认他是1MB，你是不是可以适当调大一些，比如调节到5MB？

### **重试机制** ###

> “retries”和“retries.backoff.ms”决定了重试机制，也就是如果一个请求失败了可以重试几次，每次重试的间隔是多少毫秒。
> 
> 这个大家适当设置几次重试的机会，给一定的重试间隔即可，比如给100ms的重试间隔。

### **持久化机制** ###

> “acks”参数决定了发送出去的消息要采用什么样的持久化策略，这个涉及到了很多其他的概念，这里就不再提了。

## ***个人总结：*** ##

Kafka调优通常可以从4个维度展开，分别是**吞吐量**、**延迟**、**持久性**和**可用性**。在具体展开这些方面之前，我想先建议用户保证客户端与服务器端版本一致。如果版本不一致，就会出现向下转化的问题。举个例子，服务器端保存高版本的消息，当低版本消费者请求数据时，服务器端就要做转化，先把高版本消息转成低版本再发送给消费者。这件事情本身就非常非常低效。很多文章都讨论过Kafka速度快的原因，其中就谈到了零拷贝技术——即数据不需要在页缓存和堆缓存中来回拷贝。

简单来说producer把生产的消息放到页缓存上，如果两边版本一致，可以直接把此消息推给Consumer，或者Consumer直接拉取，这个过程是不需要把消息再放到堆缓存。但是你要做向下转化或者版本不一致的话，就要额外把数据再堆上，然后再放回到Consumer上，速度特别慢。

### 1.Kafka调优 – 吞吐量 ###

>     调优吞吐量就是我们想用更短的时间做更多的事情。这里列出了客户端需要调整的参数。producer是把消息放在buffer缓存区，后端Sender线程从缓存区拿出来发到broker。这里面涉及到一个打包的过程，它是批处理的操作，不是一条一条发送的。因此这个包的大小就和TPS（系统吞吐量）息息相关。通常情况下调大这个值都会让TPS（系统吞吐量）提升，但是也不会无限制的增加。不过调高此值的劣处在于消息延迟的增加。除了**1）**调整batch.size，**2）**设置压缩也可以提升TPS（系统吞吐量），它能够减少网络传输IO。当前Lz4的压缩效果是最好的，如果客户端机器CPU资源很充足那么建议开启压缩。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 2][]

>     对于消费者端而言，调优TPS（系统吞吐量）并没有太好的办法，能够想到的就是调整fetch.min.bytes。适当地增加该参数的值能够提升consumer端的TPS。对于Broker端而言，通常的瓶颈在于副本拉取消息时间过长，因此可以适当地增加num.replica.fetcher值，利用多个线程同时拉取数据，可以加快这一进程。

### 2.Kafka调优 – 延时 ###

>     所谓的延时就是指消息被处理的时间。某些情况下我们自然是希望越快越好。针对这方面的调优，consumer端能做的不多，简单保持fetch.min.bytes默认值即可，这样可以保证consumer能够立即返回读取到的数据。说到这里，可能有人会有这样的疑问：TPS和延时不是一回事吗？假设发一条消息延时是2ms，TPS自然就是500了，因为一秒只能发500消息，其实这两者关系并不是简单的。因为我发一条消息2毫秒，但是如果把消息缓存起来统一发，TPS会提升很多。假设发一条消息依然是2ms，但是我先等8毫秒，在这8毫秒之内可能能收集到一万条消息，然后我再发。相当于你在10毫秒内发了一万条消息，大家可以算一下TPS是多少。事实上，Kafka producer在设计上就是这样的实现原理。

### 3.Kafka调优 –消息持久性 ###

>     消息持久化本质上就是消息不丢失。Kafka对消息不丢失的承诺是有条件的。以前碰到很多人说我给Kafka发消息，发送失败，消息丢失了，怎么办？严格来说Kafka不认为这种情况属于消息丢失，因为此时消息没有放到Kafka里面。Kafka只对已经提交的消息做有条件的不丢失保障。
> 
>     如果要调优持久性，对于producer而言，首先要设置**重试****以防止因为网络出现瞬时抖动造成消息发送失败**。一旦开启了重试，还需要防止乱序的问题。比如说我发送消息1与2，消息2发送成功，消息1发送失败重试，这样消息1就在消息2之后进入Kafka，也就是造成乱序了。如果用户不允许出现这样的情况，那么还需要显式地设置max.in.flight.requests.per.connection为1。
> 
> 官网解释：
> 
> max.in.flight.requests.per.connection： 客户端在阻止之前将在单个连接上发送的最大未确认请求数。请注意，如果此设置设置为大于1并且发送失败，则存在由于重试而导致消息重新排序的风险（即，如果启用了重试）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 3][]

上图列出的其他参数都是很常规的参数，比如unclean.leader.election.enable参数，最好还是将其设置成false，即不允许“脏”副本被选举为leader。

### 4.Kafka调优 –可用性 ###

> 最后是可用性，与刚才的持久性是相反的，我允许消息丢失，只要保证系统高可用性即可。因此我需要把consumer心跳超时设置为一个比较小的值，如果给定时间内消费者没有处理完消息，该实例可能就被踢出消费者组。我想要其他消费者更快地知道这个决定，因此调小这个参数的值。

\------------------------------------------------------------------------------------------------------------------------------------------------------------------------

考虑一千次，不如去做一次；犹豫一万次，不如实践一次；华丽的跌倒，胜过无谓的彷徨，将来的你，一定会感谢现在奋斗的你。欢迎大家加入大数据交流群：725967421     一起交流，一起进步！！

\------------------------------------------------------------------------------------------------------------------------------------------------------------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70]: /images/20211105/77c5ca5255ea4e3581664fb6d860942a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 1]: /images/20211105/56ec7c63b8ff4318a383205cc6ca913c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 2]: /images/20211105/d761277b6b7046e992dbdc668b8e57e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6ejExNDE5Nzk1NzA_size_16_color_FFFFFF_t_70 3]: /images/20211105/caacf245f20e4c59bbd75fc469701607.png