kafka生产者性能相关的参数理解
batch.size当多条消息发送到batch后,batch的上限大小(字节为单位),当达到上限时则会发送给kafka broker,设置的过小则降低了kafka的吞吐量,设置的过大则消息的延迟时间会拉长、占用了过高的内存...
本文主要针对生产者中几个比较重要的、常用的与性能相关的参数进行粗略的解释
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);
ProducerRecord<String, String> record = new ProducerRecord("test_topic", 1,"1", Arrays.toString(new byte[1024*1024*2]));
超过了服务器的限制
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):生产者会等待所有的副本都写入成功后才认为此消息发送成功,只要至少有一个同步副本保持活跃状态,记录就不会丢失,这是最安全的保障,是吞吐量最低的。
更多推荐
所有评论(0)