当前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秒。 

Logo

鸿蒙生态一站式服务平台。

更多推荐