Kafka中的fetch-min-size、fetch-max-wait和request.timeout.ms配置

当前kafka的版本为2.8.11，Spring Boot的版本为2.7.6，在pom.xml中引入下述依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.8.11</version>
</dependency>

然后在yml配置文件进行如下配置：

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      group-id: 0
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      fetch-max-wait: 25000ms
      fetch-min-size: 10
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

1、fetch-min-size

fetch-min-size表示消费者一次拉取请求中，服务器应该返回的最小数据量，单位是字节，默认值为1个字节。如果服务器没有足够的数据返回，请求会等待。

在Spring Boot与Kafka整合中，yml文件中的fetch-min-size参数对应着kafka的fetch.min.bytes参数。

UTF-8编码下一个字符占用一个字节

2、fetch-max-wait

fetch-max-wait表示消费者的 fetch（拉取）请求发给broker后，在broker中可能会被阻塞的（当topic中records的总size小于fetch.min.bytes时），此时这个fetch请求耗时就会比较长。这个配置就是来配置消费者最多等待response多久。

若是不满足fetch.min.bytes时，该参数就会生效，其值表示等待消费端请求的最长等待时间，默认是500ms。

在项目中创建一个生产者用于往主题 **topic0 **中投递消息，如下所示：

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaProducer {

    // 自定义的主题名称
    public static final String TOPIC_NAME="topic0";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/send")
    public String send(@RequestParam("msg")String msg) {
        log.info("准备发送消息为：{}",msg);
        // 1.发送消息
        ListenableFuture<SendResult<String,String>> future=kafkaTemplate.send(TOPIC_NAME,msg);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 2.发送失败的处理
                log.error("生产者 发送消息失败："+throwable.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, String> stringObjectSendResult) {
                // 3.发送成功的处理
                log.info("生产者 发送消息成功："+stringObjectSendResult.toString());
            }
        });
        return "接口调用成功";
    }
}

接着再在项目中创建一个消费者用于消息主题 **topic0 **中的消息，如下所示：

import java.util.Optional;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class KafkaConsumer {

    // 自定义主题名称，这里要注意的是主题名称中不能包含特殊符号：“.”、“_”
    public static final String TOPIC_NAME="topic0";

    @KafkaListener(topics = TOPIC_NAME, groupId = "ONE")
    public void topic_one(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            Object msg = message.get();
            log.info("消费者组One消费了消息：Topic:" + topic + ",Record:" + record + ",Message:" + msg);
        }
    }
}

项目启动以后，这时控制台中会打印下述信息：

ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = latest
bootstrap.servers = [127.0.0.1:9092]
client.id = consumer-ONE-1
enable.auto.commit = false
fetch.max.wait.ms = 25000
fetch.min.bytes = 10
group.id = ONE
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

接着利用** /kafka/send?msg=xxx 接口连续往主题 topic0 中生产 9 条消息，我们会发现 9 **条消息发送完了以后消费者并不会立马就可以消费到这些数据，大约在等待 **25 秒时长以上才能消费到这些数据。然后后面我们再调用/kafka/send?msg=xxx 接口连续往主题 topic0 **中生产多条消息，不管你怎么发送数据，这时消费者都可以立马消费到这些数据。

重启项目利用生产者往主题 **topic0 中投递 10 条消息，我们会发现 10 **条消息发送完了以后消费者并不会立马就可以消费到这些数据，大约在 **25 秒时长以内就能消费到这些数据。接着我们再调用/kafka/send?msg=xxx 接口连续往主题 topic0 **中生产多条消息，不管你怎么发送数据，这时消费者也是都可以立马消费到这些数据。

通过上述示例我们可以看出fetch-min-size和fetch-max-wait这两个参数的配置项确实生效了，fetch-min-size值的设置仅仅是在项目启动以后Kafka客户端第一次发起Fetch请求时（不管消费者有没有获取到数据）才会生效，仅生效一次。一旦完成一次Fetch请求后续对于消息的消费都是即时消费。而fetch-max-wait的值是一直会生效的，kefka客户端每发起一次Fetch请求，在服务端没有足够数据返回时，该Fetch请求会处于等待状态，直到达到最长等待时间。

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

如果在fetch.max.wait.ms指定的时间内，数据量依然没有达到fetch.min.bytes所设置的值，那broker也不会再等了，将直接返回数据给Consumer，此值默认为500ms。

3、request.timeout.ms

这个是生产者和消费者的通用参数配置，该配置控制客户端等待请求响应的最长时间。如果在超时时间过去之后客户端仍然未收到响应，则客户端将在必要时重新发送请求，或者在重试次数用尽时使请求失败，默认值为30秒。

例如将上述yml配置文件修改为如下所示，**fetch-max-wait **的值为31秒（超过了30秒）：

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      group-id: 0
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      fetch-max-wait: 31000ms
      fetch-min-size: 10
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

那么在项目启动以后，不管客户端有没有获取到消息，控制台中每隔一段时间就会报下述提示信息：

[Consumer clientId=consumer-ONE-1, groupId=ONE] Disconnecting from node 1001 due to request timeout.
[Consumer clientId=consumer-ONE-1, groupId=ONE] Cancelled in-flight FETCH request with correlation id 21 due to node 1001 being disconnected (elapsed time since creation: 30015ms, elapsed time since send: 30011ms, request timeout: 30000ms)
[Consumer clientId=consumer-ONE-1, groupId=ONE] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 1001:

意思就是说由于客户端和服务端的连接被断开了，取消客户端的在线fetch请求，所以一般情况下，**request.timeout.ms **的默认值未改变时，建议 **fetch-max-wait **这个值不要超过30秒。