Kafka重要生产参数配置建议

文章目录

一、消费者参数配置

1. max.poll.records

单次poll()的调用可返回的最大消息总数，

默认是500条

循环拉取并放入List集合就返回，可以看出这个取值的大小，将会影响一次poll()所需消耗的时间。

privateList<ConsumerRecord<K,V>>fetchRecords(int maxRecords){// Error when fetching the next record before deserialization.if(corruptLastRecord)thrownewKafkaException("Received exception when fetching the next record from "+ partition
                                     +". If needed, please seek past the record to "+"continue consumption.", cachedRecordException);if(isConsumed)returnCollections.emptyList();List<ConsumerRecord<K,V>> records =newArrayList<>();try{for(int i =0; i < maxRecords; i++){// Only move to next record if there was no exception in the last fetch. Otherwise we should// use the last record to do deserialization again.if(cachedRecordException ==null){
                corruptLastRecord =true;
                lastRecord =nextFetchedRecord();
                corruptLastRecord =false;}if(lastRecord ==null)break;
            records.add(parseRecord(partition, currentBatch, lastRecord));
            recordsRead++;
            bytesRead += lastRecord.sizeInBytes();
            nextFetchOffset = lastRecord.offset()+1;// In some cases, the deserialization may have thrown an exception and the retry may succeed,// we allow user to move forward in this case.
            cachedRecordException =null;}}catch(SerializationException se){
        cachedRecordException = se;if(records.isEmpty())throw se;}catch(KafkaException e){
        cachedRecordException = e;if(records.isEmpty())thrownewKafkaException("Received exception when fetching the next record from "+ partition
                                         +". If needed, please seek past the record to "+"continue consumption.", e);}return records;}

2. max.poll.interval.ms

这个参数就规定了，当调用poll()之后，如果在

max.poll.interval.ms

指定的时间内未消费完消息，也就是未再调用poll()方法，则Consumer会主动发起离开组的请求，从而产生Rebalance。

因此

max.poll.interval.ms

参数是会对Rebalance产生影响的，

默认是5分钟

，可根据实际消费能力适当调整。

3. session.timeout.ms

这个参数是用于检测当多长时间内没有收到心跳，就认为其是有故障的，从而将其中Group中移除，产生Reblance，需要注意，该值必须在broker配置

group.min.session.timeout.ms

与

group.max.session.timeout.ms

的范围内。

显然

session.timeout.ms

参数也会对Rebalance产生影响，其

默认值是10秒

。

4. heartbeat.interval.ms

Consumer的心跳频率，设置的越小频率就越高，对带宽的消耗也就越大，相反则不能快速发现实例是否已经“挂”了，通常情况下建议

heartbeat.interval.ms

设置为

session.timeout.ms

的三分之一，

默认是3秒

，如果

session.timeout.ms

有调整，记得

heartbeat.interval.ms

也需要一起调整。

5. enable.auto.commit

这个参数表示Consumer会自动定期的提交位移，

默认为自动提交

，自动提交虽然方便，但如果控制不好，很容易造成消息丢失、或者消息重复，所以一般建议改为手动提交。

6. auto.commit.interval.ms

结合

enable.auto.commit

开启自动提交，

auto.commit.interval.ms

就用来控制自动提交的频率，

默认是5秒

if(autoCommitEnabled)this.nextAutoCommitDeadline = time.milliseconds()+ autoCommitIntervalMs;privatevoidmaybeAutoCommitOffsetsAsync(long now){if(autoCommitEnabled){if(coordinatorUnknown()){this.nextAutoCommitDeadline = now + retryBackoffMs;}elseif(now >= nextAutoCommitDeadline){this.nextAutoCommitDeadline = now + autoCommitIntervalMs;doAutoCommitOffsetsAsync();}}}

7. fetch.min.bytes

这是一个调优相关的参数，

默认为1字节

，表示如果请求的数据量小于1字节，broker就是攒一攒，等足够1字节了再一起返回给Consumer，这个值建议可以适当调大一点，以提高服务的吞吐量。

8. fetch.max.wait.ms

如果在

fetch.max.wait.ms

指定的时间内，数据量依然没有达到

fetch.min.bytes

所设置的值，那broker也不会再等了，将直接返回数据给Consumer，此值

默认为500ms

二、生产者参数配置

1. batch.size

当多条消息被发送到同一个分区时，生产者会尝试把多条消息变成批量发送。这有助于提高客户端和服务器的性能。此配置以字节为单位设置默认批处理大小。如果消息大于此配置的大小，将直接发送。发送到broker的请求将包含多个批处理，每个分区一个批处理，其中包含可发送的数据。
此参数

默认值为：16KB

。
如果此参数值设置的太小，可能会降低吞吐量（批量大小为零将完全禁用批处理）。
如果此参数设置的太大，可能会更浪费内存，并增加消息发送的延迟时间。

2. linger.ms

这个参数一般会配合

batch.size

一起使用，可以通过设置

linger.ms

的值来表示，如果消息的大小一直达不到

batch.size

设置的值，那么等待多久后任然允许发送消息，

默认是不等待

，即消息到来就发送。
当我们发送的消息都比较小的时候，可以通过设置linger.ms来减少请求的次数，批次中累积更多的消息后再发送，提高了吞吐量，减少了IO请求。
如果设置的太大，则消息会被延迟更长的时间发送。

3. max.request.size

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

此参数

默认值为：1M

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

properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,1024*1024*10);

ProducerRecord<String,String> record =newProducerRecord("test_topic",1,"1",Arrays.toString(newbyte[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 withthe 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 withthe max.request.size configuration.

4. retries

生产者发送时如果遇到的是可重试的异常时，则可进行发送的重试，此参数规定了重试的次数，

默认值为Integer.MAX_VALUE

。
需要注意如果

max.in.flight.requests.per.connection

设置大于1，将有可能造成同一个分区内的消息顺序颠倒，因为如果两个批被发送到一个分区，第一批失败并重试，但第二批成功，那么第二批中的消息就有可能先出现。

5. max.in.flight.requests.per.connection

一个消息发送后在得到服务端响应之前，生产者还可以发送的消息条数，配合

retries

使用，可以保证消息的顺序性，假设有两条消息A、B，A先发送但失败了在执行重试时，B发送且成功了，之后A也重试成功了，此时A、B消息顺序就反了，如果将此参数设置为1，则可以保证A在重试时，B消息无法进行发送，必须等A收到broker响应后B才能发送，设置较高可以提升吞吐量，但会占用更多的内存，此参数值

默认是5条

。

6. acks

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

这个参数一共有3个值，

默认为1

。

0：生产者只要把消息发送出去即可，不用等待broker的处理结果，消息将立即添加到socket buffer并被视为已发送。在这种情况下，无法保证服务器已收到消息，并且

retries

配置将不会生效（因为客户端通常不会知道任何故障）。为每条消息返回的偏移量将始终设置为-1。
设置为0，吞吐量最高，同样消息的丢失率也最高。
1：生成者需要等分区leader将消息写入成功后才认为此消息发送成功，兼顾了吞吐量和消息丢失的问题，但是同样有消息丢失的风险，比如当leader写入成功后突然挂了，其他分区跟随者并为能够将此消息同步，则此消息丢失。
all（等同于-1）：生产者会等待所有的副本都写入成功后才认为此消息发送成功，只要至少有一个同步副本保持活跃状态，消息就不会丢失，这是最安全的保障，是吞吐量最低的。

7. buffer.memory

生产者可用于缓冲等待发送到服务器的记录的内存总字节数，如果客户端send的速度大于发送到broker的速度，且积压的消息大于这个设置的值，就会造成send阻塞，阻塞时间为

max.block.ms

设置的值，如果超过时间就抛出异常。

8. max.block.ms

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

默认是60秒

。

指定把消息发送到一个不存在的分区，模拟获取元数据异常，设置阻塞时间为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.

三、生产者、消费者通用参数配置

1. send.buffer.bytes和receive.buffer.bytes

send.buffer.bytes和receive.buffer.bytes

分别指定了发送数据和接收数据时要使用的TCP发送缓冲区(SO_SNDBUF)和接收缓冲区(SO_RCVBUF)的大小，

send.buffer.bytes默认是128KB，receive.buffer.bytes默认是32KB

，如果指定为-1，则表示使用操作系统默认大小。

2. request.timeout.ms

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

默认值为：30秒

。

3. retry.backoff.ms

主要用来控制请求发送异常后

retries

的请求频率，

默认是100ms

。

4. metadata.max.age.ms

Kafka客户端定期刷新元数据的频率，

默认是5分钟

。

四、Broker参数配置

1. auto.create.topics.enable

是否允许自动创建topic，

默认为自动创建

，一般建议设置为不允许自动创建，否则可能会生成很多无效的Topic，造成资源浪费，也增加了运维成本。

2. unclean.leader.election.enable

这个参数表明了什么样的副本才有资格竞争Leader，如果设置成false，则表示对于落后太多的副本是没有资格竞选Leader的，这样做的后果是可能永远选不出Leader来了，如果是true，那么则有可能选出一个差很多数据的副本成为Leader，从而造成了数据丢失，此值

默认是false

，建议也设置成false。

3. replica.lag.time.max.ms

这个参数就定义了上面副本落后的标准，其表示Follower副本能够落后Leader副本的最长时间间隔，

默认值为10秒

，只要不超过10秒，就不算是落后的副本。

4. auto.leader.rebalance.enable

如果把

auto.leader.rebalance.enable

设置为true，则Kafka会定期的对Topic的分区进行Leader选举，他很有可能会莫名其妙的把原本好好的Leader换掉，此参数值

默认为true

，一般建议改为false。

5. message.max.bytes

这个参数表示Broker能够处理的最大消息大小，

默认为1M

，建议适当调大一些，一般这个参数还会结合

max.message.bytes

、

replica.fetch.max.bytes

以及Consumer端的

fetch.message.max.bytes

一起使用。

6. replication.factor

这个参数用来表示分区的副本数，建议大于等于3，确保Kafka的高可用性，

默认是1

。

7. min.insync.replicas

这个参数定义了消息至少要被写入多少个副本才算是“已提交”，

默认是1

，建议设置成大于1。
注意：

min.insync.replicas

只有在

ack

设置成all(-1)时才会生效。

8. num.io.threads

表示每台Broker启动后自动创建的I/O线程数量，I/O线程是真正处理请求的线程，

默认是8个

。

9. num.network.threads

表示每台Broker启动时专门用于从网络接收请求并向网络发送响应的线程数量，

默认是3个

。

Kafka会通过使用网络线程来专门接收客户端的请求与和发送响应，但是当收到请求之后网络线程并不是真正的由自己来处理，而是丢到一个共享请求队列中，此时就会由另一个线程池中的线程来处理，也就是I/O线程，它会专门负责从共享队列中取出请求，执行真正的处理。