1. kafka-提高生产者吞吐量

news2024/11/16 3:49:08

kafka的基础使用知识不在这篇文章阐述，默认读者都会使用

首先kafka的生产者原理，如下。图来自网上
在这里插入图片描述
上面的图可能不全，我再详细描述下，在消息发送过程中，设计到两个线程main和sender线程；

（1）main线程会创建一个RecordAccumulator的双端队列（可以理解为缓冲区）
在这里插入图片描述
main线程不断发送消息到缓冲区，而sender线程不断从缓冲区拿消息发送到Broker中

（2）在发送过程中，其中会经过三个步骤，a-拦截器（使用场景不多），b-序列化器，c-Partioner（分区器）

（3）经过三个这些步骤后，消息会放到缓冲区中，但是kafka不会来一个消息就发一个到broker中，而是根据batch(批次)来，当以下任意一个条件满足，就会发送消息（主要流程，还有一些别逻辑）
在这里插入图片描述
翻译下就是：
a - batch满了。（默认16k）
b - 等待时间到了（lingerMS默认是0ms，这个真要设置，不然来一个发一个，吞吐差）
c - RecordAccumulator满了。（默认32mm，这情况所有分区Partioner立马ready好，等待发送）
d - RecordAccumulator关闭了

（4）所以结论来说，提高kafka生产者的吞吐性能，涉及以下几个参数
batch-size、linger.ms、缓冲区大小和消息压缩（消息被压缩了，一次就能发送更多消息，带来更好的吞吐性能）

简化代码案例

package cn.sysu.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerParameters {


	public static void main(String[] args) {
		// 1.配置参数
		Properties properties = new Properties();
		// 1.1连接kafka集群
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
		// 1.2序列化
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		// 1.3根据业务场景，提高生产者的吞吐量
		properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,32 * 1024 * 1024);//缓冲区大小
		properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16* 1024);//批次大小
		properties.put(ProducerConfig.LINGER_MS_CONFIG,5); //Linger.ms
		properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy"); //压缩


		// 2.创建生产者
		KafkaProducer<String,String> kafkaProducer = new KafkaProducer<>(properties);
		// 3.发送消息
		for (int i = 0; i < 100; i++) {
			kafkaProducer.send(new ProducerRecord<>("topic","消息体"+i));
		}

		// 4.关闭资源，根据实际生产环境，可能生产者生命周期与服务一样
		kafkaProducer.close();
	}

}