kafka性能调优（生产环境）

一、前言前段时间，公司项目测试，测试组发现kafka topic数据与发送量不一致，后来查看日志发现报了这个错误。Got error produce response with correlation id xxx on topic-partition xxxxx, retrying (9 attempts left). Error: NETWORK_EXCEPTIONGot error prod

rookie`

15111人浏览 · 2021-04-20 10:23:48

rookie` · 2021-04-20 10:23:48 发布

一、前言

前段时间，公司项目测试，测试组发现kafka topic数据与发送量不一致，后来查看日志发现报了这个错误。

Got error produce response with correlation id xxx on topic-partition xxxxx, retrying (9 attempts left). Error: NETWORK_EXCEPTION
Got error produce response with correlation id xxx on topic-partition xxxxx, retrying (9 attempts left). Error: REQUEST_TIMED_OUT

这个问题，通过查询官网以及论坛搜索发现是由于kafka压力过大，从而导致的网络异常，最终导致kafka数据丢失。

二、kafka版本

CDH6.0.0+1.0.0Kafka+3.4.5Zookeeper

三、性能调优

1)num.network.threads=cpu核数+1 #网络通信处理线程数
kafka主机参数
在这里插入图片描述
2)num.io.threads=cpu核数*2 #io磁盘io处理线程数

3)replica.lag.time.max.ms= 60000# 如果网络不好,或者kafka集群压力较大,会出现副本丢失,然后会频繁复制副本,导致集群压力更大,此时可以调大该参数。

4)compression.type:none gzip snappy lz4
#默认发送不进行压缩，推荐配置一种适合的压缩算法，可以大幅度的减缓网络压力和Broker的存储压力。
在这里插入图片描述
5)kafka默认副本数，default.replication.factor = 2（建议为2，或3）

6)ISR队列最小副本数，min.insync.replicas = 2（建议为2）

此参数配合produce端，“acks” = “-1"使用，保证数据不丢失。必须配置两个参数都满足。
7）ISR队列副本同步最大等待时间，超过此时间未向leader同步数据的副本将被踢出ISR队列 ,replica.lag.time.max.ms = 30000
在这里插入图片描述
8)Kafka内存调整KAFKA_HEAP_OPTS=”-Xms4g -Xmx4g"

9)batch.size 设置，可适当调大。每批次拉取的数据量，等待拉取的数据量达到这个值，才发往kafka。配合linger.ms 参数使用，没有达到配置量，超过最大等待时间，也会发送kafka。
10)kafka生产者配置，以下给出案例参数

 private static Properties props;
    static {
        props = new Properties();
        //kafka集群，broker-list
        props.put("bootstrap.servers", "xx.xxx.xxx.xx:9092,xx.xxx.xxx.xx:9092,xx.xxx.xxx.xx:9092");
        //ack设置为-1/all
        props.put("acks", "all");
        //重试次数
        props.put("retries", Integer.MAX_VALUE);
        //批次大小
        props.put("batch.size", 163840);
        //等待时间
        props.put("linger.ms", 50);
        //RecordAccumulator缓冲区大小
        props.put("buffer.memory", 67108864);
        //单条消息最大大小
        props.put("max.request.size", 10485760);
        //失败重试间隔
        props.put("retry.backoff.ms", 500);
        //kafka消息序列化，反序列化key、value设置
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }