Kafka 生产调优

Kafka生产调优

一、Kafka 硬件配置选择

场景说明

100 万日活，每人每天 100 条日志，每天总共的日志条数是 100 万 * 100 条 = 1 亿条。

1 亿 / 24 小时 / 60 分 / 60 秒 = 1150 条/每秒钟。

每条日志大小：0.5k ~ 2k（约1k）。

1150 条/每秒钟 * 1k ≈ 1m/s 。

高峰期每秒钟：1150 条 * 20 倍 = 23000 条。

所以高峰每秒数据量：20MB/s。

服务器台数选择

服务器台数= 2 * （生产者峰值生产速率 * 副本 / 100） + 1

​ = 2 * （20m/s * 2 / 100） + 1

​ = 3 台

磁盘选择

kafka 底层主要是顺序写，固态硬盘和机械硬盘的顺序写速度差不多。建议选择普通的机械硬盘。

每天总数据量：1 亿条 * 1k ≈ 100g

100g * 副本 2 * 保存时间 3 天 / 0.7 ≈ 1T

建议三台服务器硬盘总大小，大于等于 1T。

内存选择

Kafka 内存组成：堆内存 + 页缓存

1）Kafka 堆内存建议每个节点：10g ~ 15g（在 kafka-server-start.sh 中修改）

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
 export KAFKA_HEAP_OPTS="-Xmx10G -Xms10G"
fi

2）页缓存：页缓存是 Linux 系统服务器的内存。我们只需要保证 1 个 segment（1g）中25%的数据在内存中就好。

每个节点页缓存大小 =（分区数 * 1g * 25%）/ 节点数。例如 10 个分区，页缓存大小=（10 * 1g * 25%）/ 3 ≈ 1g

建议服务器内存大于等于 11G。

CPU选择

• num.io.threads = 8 负责写磁盘的线程数，整个参数值要占总核数的 50%。
• num.replica.fetchers = 1 副本拉取线程数，这个参数占总核数的 50%的 1/3。
• num.network.threads = 3 数据传输线程数，这个参数占总核数的 50%的 2/3。

建议 32 个 cpu core。

二、Kafka Broker调优

Broker 核心参数配置

服役新节点/退役旧节点

（1）创建一个要均衡的主题。

$ vim topics-to-move.json 
{
   "topics": [
 		{"topic": "first"}
 	],
 	"version": 1
}

（2）生成一个负载均衡的计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate

（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中），由步骤2生成的👆

$ vim increase-replication-factor.json

（4）执行副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --execute

（5）验证副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --verify

增加副本因子

1）创建测试 topic（3分区、1副本）

$ bin/kafka-topics.sh --bootstrap-server node102:9092 --create --partitions 3 --replication-factor 1 --
topic four

2）手动增加副本存储（3分区、3副本），创建副本存储计划，所有副本都指定存储在 broker0、broker1、broker2 中

vim increase-replication-factor.json

{"version":1,"partitions":[
	{"topic":"four","partition":0,"replicas":[0,1,2]},
	{"topic":"four","partition":1,"replicas":[0,1,2]},
	{"topic":"four","partition":2,"replicas":[0,1,2]}
]}

3）执行副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --execute

调整分区副本存储

（1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。

 vim increase-replication-factor.json

{
	"version":1,
	"partitions":[{"topic":"three","partition":0,"replicas":[0,1]},
	{"topic":"three","partition":1,"replicas":[0,1]},
	{"topic":"three","partition":2,"replicas":[1,0]},
	{"topic":"three","partition":3,"replicas":[1,0]}
	]
}

（2）执行副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --execute

（3）验证副本存储计划

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --verify

三、Kafka 生产者调优

生产者如何提高吞吐量

数据可靠性

至少一次 = ACK 级别设置为-1 + 分区副本大于等于 2 + ISR 里应答的最小副本数量大于等于 2

数据去重

Kafka 的事务一共有如下 5 个 API

// 1 初始化事务
void initTransactions();

// 2 开启事务
void beginTransaction() throws ProducerFencedException;

// 3 在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,String consumerGroupId) throws 
ProducerFencedException;

// 4 提交事务
void commitTransaction() throws ProducerFencedException;

// 5 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;

数据乱序

四、Kafka 消费者调优

消费者重要参数🚩

消费者再平衡

指定offset进行消费

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringDeserializer");
   properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                   "org.apache.kafka.common.serialization.StringDeserializer");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   // 2 订阅一个主题
   ArrayList<String> topics = new ArrayList<>();
   topics.add("first");
   kafkaConsumer.subscribe(topics);
   
   //🚩 获取消费者分区信息，并指定offset进行消费
   Set<TopicPartition> assignment= new HashSet<>();
 	  while (assignment.size() == 0) {
 		 kafkaConsumer.poll(Duration.ofSeconds(1));
 		 // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
 		 assignment = kafkaConsumer.assignment();
       }
 	    
      // 遍历所有分区，并指定 offset 从 1700 的位置开始消费
	  for (TopicPartition tp: assignment) {
	  	kafkaConsumer.seek(tp, 1700);
	  }
     
     // 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       		// 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 	}
     
  }
}   

指定时间进行消费

在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。例如要求按照时间消费前一天的数据，怎么处理？

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringDeserializer");
   properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                   "org.apache.kafka.common.serialization.StringDeserializer");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   // 2 订阅一个主题
   ArrayList<String> topics = new ArrayList<>();
   topics.add("first");
   kafkaConsumer.subscribe(topics);
   
   //🚩 获取消费者分区信息，并指定offset进行消费
   Set<TopicPartition> assignment= new HashSet<>();
   while (assignment.size() == 0) {
 	  kafkaConsumer.poll(Duration.ofSeconds(1));
 	  // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
 	  assignment = kafkaConsumer.assignment();
   }
     
   HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>();	    
   // 封装集合存储，每个分区对应一天前的数据
   for (TopicPartition topicPartition : assignment) {
       timestampToSearch.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
   }
    // 获取从 1 天前开始消费的每个分区的 offset
 	Map<TopicPartition, OffsetAndTimestamp> offsets = kafkaConsumer.offsetsForTimes(timestampToSearch);
    // 遍历每个分区，对每个分区设置消费时间。
 	for (TopicPartition topicPartition : assignment) {
 		OffsetAndTimestamp offsetAndTimestamp = offsets.get(topicPartition);
 		// 根据时间指定开始消费的位置
 		if (offsetAndTimestamp != null){
 			kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset());
 		}
 	}
     
     // 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       		// 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 	}
     
  }
}   	

消费者如何提高吞吐量

增加分区数：

$ bin/kafka-topics.sh --bootstrap-server node102:9092 --alter --topic first --partitions 3

五、Kafka总体调优

如何提升吞吐量🚩

1）提升生产吞吐量

• buffer.memory：发送消息的缓冲区大小，默认值是 32m，可以增加到 64m。
• batch.size：默认是 16k。如果 batch 设置太小，会导致频繁网络请求，吞吐量下降；如果 batch 太大，会导致一条消息需要等待很久才能被发送出去，增加网络延时
• linger.ms，这个值默认是 0，意思就是消息必须立即被发送。一般设置一个 5~100毫秒。如果 linger.ms 设置的太小，会导致频繁网络请求，吞吐量下降；如果 linger.ms 太长，会导致一条消息需要等待很久才能被发送出去，增加网络延时。
• compression.type：默认是 none，不压缩，但是也可以使用 lz4 压缩，效率还是不错的，压缩之后可以减小数据量，提升吞吐量，但是会加大 producer 端的 CPU 开销。

2）增加分区

3）消费者提高吞吐量

• 调整 fetch.max.bytes 大小，默认是 50m。
• 调整 max.poll.records 大小，默认是 500 条。

4）增加下游消费者处理能力

数据精准一次

1）生产者角度

• acks 设置为-1 （acks=-1）
• 幂等性（enable.idempotence = true） + 事务

2）broker 服务端角度

• 分区副本大于等于 2 （–replication-factor 2）
• ISR 里应答的最小副本数量大于等于 2 （min.insync.replicas = 2）

3）消费者

• 事务 + 手动提交 offset （enable.auto.commit = false）
• 消费者输出的目的地必须支持事务（MySQL、Kafka）

合理设置分区数

（1）创建一个只有 1 个分区的 topic。

（2）测试这个 topic 的 producer 吞吐量和 consumer 吞吐量。

（3）假设他们的值分别是 Tp 和 Tc，单位可以是 MB/s。

（4）然后假设总的目标吞吐量是 Tt，那么分区数 = Tt / min（Tp，Tc）。

例如：producer 吞吐量 = 20m/s；consumer 吞吐量 = 50m/s，期望吞吐量 100m/s；

分区数 = 100 / 20 = 5 分区

分区数一般设置为：3-10 个

分区数不是越多越好，也不是越少越好，需要搭建完集群，进行压测，再灵活调整分区个数。

单条日志大于 1m的问题

集群压力测试

用 Kafka 官方自带的脚本，对 Kafka 进行压测

• 生产者压测：kafka-producer-perf-test.sh
• 消费者压测：kafka-consumer-perf-test.sh

Kafka Producer 压力测试

（1）创建一个 test topic，设置为 3 个分区 3 个副本

$ bin/kafka-topics.sh --bootstrap-server node102:9092 --create --replication-factor 3 --partitions 3 --topic test

（2）在/opt/module/kafka/bin 目录下面有这两个文件。我们来测试一下

$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=node102:9092,node103:9092,node104:9092 batch.size=16384 linger.ms=0

测试参数说明：

• record-size 是一条信息有多大，单位是字节，本次测试设置为 1k。
• num-records 是总共发送多少条信息，本次测试设置为 100 万条。
• throughput 是每秒多少条信息，设成-1，表示不限流，尽可能快的生产数据，可测出生产者最大吞吐量。本次设置为每秒钟 1 万条。
• producer-props 后面可以配置生产者相关参数，batch.size 配置为 16k

调优参数：

更改不同的值进行压测

Kafka Consumer 压力测试

（1）修改/opt/module/kafka/config/consumer.properties 文件中的一次拉取条数为 500

max.poll.records=500

（2）消费 100 万条日志进行压测

$ bin/kafka-consumer-perf-test.sh --bootstrap-server node102:9092,node103:9092,hadoop104:9092 --topic test --messages 1000000 --consumer.config config/consumer.properties

参数说明：

• –bootstrap-server 指定 Kafka 集群地址
• –topic 指定 topic 的名称
• –messages 总共要消费的消息个数。本次实验 100 万条

调优参数优化：（在consumer.properties中进行修改）

• 修改一次拉取的数量：max.poll.records=2000
• 调整文件中的拉取一批数据大小 100m：fetch.max.bytes=104857600