kafka生产者性能相关的参数理解_kafka batch.size

本文主要针对生产者中几个比较重要的、常用的与性能相关的参数进行粗略的解释

batch.size
默认值为16KB
当多条消息被发送到同一个分区时，生产者会尝试把多条消息变成批量发送。这有助于提高客户端和服务器的性能。此配置以字节为单位设置默认批处理大小。如果消息大于此配置的大小，将直接发送。发送到broker的请求将包含多个批处理，每个分区一个批处理，其中包含可发送的数据。
如果此参数值设置的太小，可能会降低吞吐量（批量大小为零将完全禁用批处理）。
如果此参数设置的太大，可能会更浪费内存，并增加消息发送的延迟时间。

linger.ms
默认值为0，即不等待
这个参数一般会配合batch.size一起使用，可以通过设置linger.ms的值来表示，如果消息的大小一直达不到batch.size设置的值，那么等待多久后任然允许发送消息，默认是不等待，即消息到来就发送。
当我们发送的消息都比较小的时候，可以通过设置linger.ms来减少请求的次数，批次中累积更多的消息后再发送，提高了吞吐量，减少了IO请求。
如果设置的太大，则消息会被延迟更长的时间发送。

public static void main(String[] args) {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "192.168.93.132:9092");
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 0);
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024 );
    KafkaProducer<String, String> producer = new KafkaProducer(properties);
    new Thread(() -> {
        ProducerRecord<String, String> record = new ProducerRecord("test_topic", 1,"1", Arrays.toString(new byte[1024]));
        Future future = producer.send(record);
        long s = System.currentTimeMillis();
        try {
            future.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        long e = System.currentTimeMillis();
        System.out.println("producer send msg wait time: " + (e - s));
    }).start();
    new Thread(() -> {
        ProducerRecord<String, String> record = new ProducerRecord("test_topic", 1,"1", Arrays.toString(new byte[1024]));
        Future future = producer.send(record);
        long s = System.currentTimeMillis();
        try {
            future.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        long e = System.currentTimeMillis();
        System.out.println("producer send msg wait time: " + (e - s));
    }).start();
}

配置batch.size为16kb，linger.ms不延迟，发送1024字节的数据，执行后结果为：

properties.put(ProducerConfig.LINGER_MS_CONFIG, 0);
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024 );

producer send msg wait time: 76
producer send msg wait time: 76

配置batch.size为16kb，linger.ms延迟1秒，发送1024字节的数据，执行后结果为：

properties.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024 );

producer send msg wait time: 1023
producer send msg wait time: 1023

配置batch.size为16字节，linger.ms不延迟，发送1024字节的数据，执行后结果为：

properties.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16);

producer send msg wait time: 38
producer send msg wait time: 49

配置batch.size为16字节，linger.ms延迟1秒，发送1024字节的数据，执行后结果为：

properties.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16);

producer send msg wait time: 8
producer send msg wait time: 1005

buffer.memory
默认值为32M
生产者可用于缓冲等待发送到服务器的记录的内存总字节数，如果客户端send的速度大于发送到broker的速度，且积压的消息大于这个设置的值，就会造成send阻塞，阻塞时间为max.block.ms设置的值，如果超过时间就抛出异常。

max.block.ms
默认值为60s
当执行KafkaProducer.send() 和KafkaProducer.partitionsFor()时阻塞等待的时间，之所以会阻塞时因为可能buffer满了或者获取元数据异常，那么超过这个时间就会抛出异常。

指定把消息发送到一个不存在的分区，模拟获取元数据异常，设置阻塞时间为1秒，报错信息如下。

properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 1000);
 
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Topic test_topic not present in metadata after 1000 ms.
    at org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:1269)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:933)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:856)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:743)
    at cn.enjoyedu.hellokafka.HelloKafkaProducer.lambda$main$0(HelloKafkaProducer.java:26)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.errors.TimeoutException: Topic test_topic not present in metadata after 1000 ms.
 

max.request.size
默认值为1M
设置请求消息的最大大小，避免发送大量的请求，限制了单条消息的size与批次消息的size，如果改变此值，需要注意服务器也需要进行相应设置，因为服务器也有接收消息的大小限制。

设置大小为10M，发送2M大小的消息

properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1024*1024*10);
1
ProducerRecord<String, String> record = new ProducerRecord("test_topic", 1,"1", Arrays.toString(new byte[1024*1024*2]));
1
超过了服务器的限制

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:98)
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:67)
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:30)
    at cn.enjoyedu.hellokafka.HelloKafkaProducer.lambda$main$0(HelloKafkaProducer.java:30)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.

设置大小为1M，发送2M大小的消息，报错超过max.request.size限制。

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.RecordTooLargeException: The message is 6291545 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
    at org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:1269)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:933)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:856)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:743)
    at cn.enjoyedu.hellokafka.HelloKafkaProducer.lambda$main$0(HelloKafkaProducer.java:27)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 6291545 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.

retries
默认值为Integer.MAX_VALUE
生产者发送时如果遇到的是可重试的异常时，则可进行发送的重试，此参数规定了重试的次数，需结合retries.backoff.ms使用（默认100ms），规定每次重试间隔的时间大小。
需要注意如果max.in.flight.requests.per.connection设置大于1，将有可能造成同一个分区内的消息顺序颠倒，因为如果两个批被发送到一个分区，第一批失败并重试，但第二批成功，那么第二批中的消息就有可能先出现。
官方建议通过这个参数delivery.timeout.ms来控制重试行为。

max.in.flight.request.per.connection
默认值为5
一个消息发送后在得到服务端响应之前，生产者还可以发送的消息条数，配合retries使用，可以保证消息的顺序性，假设有两条消息A、B，A先发送但失败了在执行重试时，B发送且成功了，之后A也重试成功了，此时A、B消息顺序就反了，如果将此参数设置为1，则可以保证A在重试时，B消息无法进行发送，必须等A收到broker响应后B才能发送，设置较高可以提升吞吐量，但会占用更多的内存，设置过高反而会降低吞吐量，因为批量消息的效率降低

delivery.timeout.ms
默认值为2分钟
调用send()返回后收到成功或失败的时间上限。这限制了消息在发送之前被延迟的总时间、等待broker确认的时间（如果可预期的话）以及允许可重试发送失败的时间。如果遇到不可恢复的错误、重试次数已用尽、消息被添加到过期批次中，则会发生异常，此配置的值应大于或等于request.timeout.ms、linger.ms时间的之和

设置delivery.timeout.ms为1秒，报错

Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to construct kafka producer
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:433)
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:298)
    at cn.enjoyedu.hellokafka.HelloKafkaProducer.main(HelloKafkaProducer.java:27)
Caused by: org.apache.kafka.common.config.ConfigException: delivery.timeout.ms should be equal to or larger than linger.ms + request.timeout.ms
    at org.apache.kafka.clients.producer.KafkaProducer.configureDeliveryTimeout(KafkaProducer.java:492)
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:393)
    ... 2 more

使用debug断点，模拟超时，设置delivery.timeout.ms为2秒，得到报错信息如下

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for test_topic-1:3288 ms has passed since batch creation
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:98)
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:67)
    at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:30)
    at cn.enjoyedu.hellokafka.HelloKafkaProducer.lambda$main$0(HelloKafkaProducer.java:33)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for test_topic-1:3288 ms has passed since batch creation

request.timeout.ms
默认值为30秒
配置控制客户端等待请求响应的最长时间。如果在超时时间过去之前未收到响应，则客户端将在必要时重新发送请求，或者在重试次数用尽时使请求失败。

acks
默认值为1

生产者在确认请求完成之前要求leader已收到的确认数。这控制了发送的消息的持久性。

这个参数一共有3个值。

0：生产者只要把消息发送出去即可，不用等待broker的处理结果，记录将立即添加到socket buffer并被视为已发送。在这种情况下，无法保证服务器已收到记录，并且retries配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量将始终设置为-1。
设置为0，吞吐量最高，同样消息的丢失率也最高
1：生成者需要等分区leader将消息写入成功后才认为此消息发送成功，兼顾了吞吐量和消息丢失的问题，但是同样有消息丢失的风险，比如当leader写入成功后突然挂了，其他分区跟随者并为能够将此消息同步，则此消息丢失。
all（等同于-1）：生产者会等待所有的副本都写入成功后才认为此消息发送成功，只要至少有一个同步副本保持活跃状态，记录就不会丢失，这是最安全的保障，是吞吐量最低的。