Kafka重要生产参数配置建议_receive.buffer.bytes

一、消费者参数配置

1. max.poll.records

单次poll()的调用可返回的最大消息总数，默认是500条

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

private List<ConsumerRecord<K, V>> fetchRecords(int maxRecords) {
    // Error when fetching the next record before deserialization.
    if (corruptLastRecord)
        throw new KafkaException("Received exception when fetching the next record from " + partition
                                     + ". If needed, please seek past the record to "
                                     + "continue consumption.", cachedRecordException);
    if (isConsumed)
        return Collections.emptyList();
    List<ConsumerRecord<K, V>> records = new ArrayList<>();
    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())
            throw new KafkaException("Received exception when fetching the next record from " + partition
                                         + ". If needed, please seek past the record to "
                                         + "continue consumption.", e);
    }
    return records;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

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;

private void maybeAutoCommitOffsetsAsync(long now) {
    if (autoCommitEnabled) {
        if (coordinatorUnknown()) {
            this.nextAutoCommitDeadline = now + retryBackoffMs;
        } else if (now >= nextAutoCommitDeadline) {
            this.nextAutoCommitDeadline = now + autoCommitIntervalMs;
            doAutoCommitOffsetsAsync();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

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);
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.
1
2
3
4
5
6
7

设置大小为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.
1
2
3
4
5
6
7
8

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);
1

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
2
3
4
5
6
7
8

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

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线程，它会专门负责从共享队列中取出请求，执行真正的处理。