ToB企服应用市场:ToB评测及商务社交产业平台

标题: Kafka中的fetch-min-size、fetch-max-wait和request.timeout.ms设置 [打印本页]

作者: 徐锦洪 时间: 2024-7-12 23:19
标题: 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秒。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)