前言:
前面我们分享了 Kafka 的一些基础知识,以及 Spring Boot 集成 Kafka 完成消息发送消费,本篇我们来分享一下 Kafka 的批量消息发送消费。
Kafka 系列文章传送门
Kafka 客户端工具使用分享【offsetexplorer】
Kafka 消息批量发送
Kafka 没有提供批量发送消息的 API,Kafka 的方式是提供一个 RecordAccumulator 消息收集器,将发送给同一个 Topic 同一个 Partition 的消息先缓存起来,当其达到某些条件后,才会一次性的将消息提交给 Kafka Broker。
Kafka 消息的批量发送主要跟以下三个参数有关:
- batch.size:批量发送消息的大小,默认 16KB,产生的消息达到这个数量后,即刻触发消息批量提交到 Kafka Broker。
- buffer.memory:生产者可用于缓冲等待发送到服务器的消息占用的总内存字节数,模式是 32 MB,如果超过这个数量,即刻触发消息批量提交到 Kafka Broker。
- linger.ms:批量发送的的最大时间间隔,单位是毫秒,当达到配置的时间之后,会立刻触发消息批量提交大 Kafka Broker。
以上三个条件满足一个就会触发消息的批量提交。
Kafka 批量消息 参数配置
上面我们分析了 Kafka 没有提供批量发送的 API,而是使用了三个参数来控制批量发送的,换句话说,其实我们每次使用 Kafka 发送消息的时候都是批量发送,Kafka 批量发送消息的代码没有什么特殊之处,只需要对上面解释的三个参数进行按需配置即可,本案例的配置如下:
#批量发送消息的大小 默认 16KB 我们这里为了演示效果 配置为1Kb
spring.kafka.producer.batch-size = 1024
#生产者可用于缓冲等待发送到服务器的消息占用的总内存字节数 默认 32M
spring.kafka.producer.buffer-memory = 33554432
#批量发送的的最大时间间隔,单位是毫秒
spring.kafka.producer.properties.linger.ms=50000
Kafka 批量消息 Producer 代码演示
Kafka 批量发送消息代码如下:
package com.order.service.kafka.producer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* @ClassName: MyKafkaBatchProducer
* @Author: Author
* @Date: 2024/10/22 19:22
* @Description:
*/
@Slf4j
@Component
public class MyKafkaBatchProducer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void batchSendMessage() {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(new Date());
log.info("开始消息发送,当前时间:{}", dateStr);
for (int a = 0; a < 1000; a++) {
this.kafkaTemplate.send("my-topic", "第" + a + "条 kafka 消息");
}
log.info("完成消息发送,当前时间:{}", dateStr);
}
}
在 Kafka 发送完成消息后,我们记录了当前时间,这个时间是用来证明消息是被批量发送的。
Kafka 批量消息 Consumer 代码演示
Kafka 批量消息的代码也没有什么特殊之处,还是使用 @KafkaListener注解来监听消息,只不过参数变成了 List<ConsumerRecord<String, String>> 类型,然后我们在配置中配置了批量消费的模式,批量消费的配置如下:
#Kafka 的消费模式 single 每次单条消费消息 batch 每次批量消费消息
spring.kafka.listener.type = batch
Consumer 代码如下:
package com.order.service.kafka.consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* @ClassName: MyKafkaBatchConsumer
* @Author: Author
* @Date: 2024/10/22 19:22
* @Description:
*/
@Slf4j
@Component
public class MyKafkaBatchConsumer {
@KafkaListener(id = "my-kafka-consumer-01",
groupId = "my-kafka-consumer-groupId-01",
topics = "my-topic",
containerFactory = "myContainerFactory",properties = {"max.poll.records:10"})
public void listen(List<ConsumerRecord<String, String>> consumerRecords) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(new Date());
log.info("my-kafka-consumer-groupId-01 消息消费成功,当前时间:{},消息size:{}", dateStr, consumerRecords.size());
for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
String value = consumerRecord.value();
log.info("消息内容:{}",value);
}
}
}
这里我们使用了 properties 这个属性配置,后面详细讲解。
** Kafka 批量消息验证**
触发消息发送消费结果如下:
2024-10-27 15:27:17.563 INFO 18320 --- [nio-8086-exec-2] c.o.s.k.producer.MyKafkaBatchProducer : 完成消息发送,当前时间:2024-10-27 15:27:17
2024-10-27 15:27:22.569 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : my-kafka-consumer-groupId-01 消息消费成功,当前时间:2024-10-27 15:27:22,消息size:10
2024-10-27 15:27:22.569 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第0条 kafka 消息
2024-10-27 15:27:22.569 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第1条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第2条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第3条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第4条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第5条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第6条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第7条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第8条 kafka 消息
2024-10-27 15:27:22.570 INFO 18320 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : 消息内容:第9条 kafka 消息
2024-10-27 15:27:17 完成消息发送,:2024-10-27 15:27:22 完成消息消费,时间间隔是 5秒,消息是 10 条,符合预期。
我们修改配置再次演示,将批量发送消息的时间间隔改为 10 秒,同时一次性发送 1000 条消息,是消息的总大小大于 1KB。
#批量发送消息的大小 默认 16KB 我们这里为了演示效果 配置为1Kb
spring.kafka.producer.batch-size = 1024
#生产者可用于缓冲等待发送到服务器的消息占用的总内存字节数 默认 32M
spring.kafka.producer.buffer-memory = 33554432
#批量发送的的最大时间间隔,单位是毫秒
spring.kafka.producer.properties.linger.ms=50000
调整消息发送端代码如下:
package com.order.service.kafka.producer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
/**
* @ClassName: KafkaProducer
* @Author: Author
* @Date: 2024/10/22 19:22
* @Description:
*/
@Slf4j
@Component
public class MyKafkaBatchProducer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void batchSendMessage() {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(new Date());
log.info("开始消息发送,当前时间:{}", dateStr);
for (int a = 0; a < 1000; a++) {
this.kafkaTemplate.send("my-topic", "第" + a + "条 kafka 消息");
}
log.info("完成消息发送,当前时间:{}", dateStr);
}
}
触发消息发送消费结果如下:
2024-10-27 15:41:39.530 INFO 17440 --- [nsumer-01-2-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : my-kafka-consumer-groupId-01 消息消费成功,当前时间:2024-10-27 15:41:39,消息size:10
2024-10-27 15:41:39.530 INFO 17440 --- [nsumer-01-0-C-1] c.o.s.k.consumer.MyKafkaBatchConsumer : my-kafka-consumer-groupId-01 消息消费成功,当前时间:2024-10-27 15:41:39,消息size:10
可以看到消息发送和消息消费几乎是同时进行的,因为这里我们打印的是时间只有秒,是看不出差异的,但是也可以根据这个结果看出,消费者并没有等到 10秒后才开始消费,是因为批量发送消息的大小大于了1KB 就触发了批量消息的提交,符合上面我们说的三个条件满足其中一个就触发批量消息提交到 Kafka Broker,结果符合预期。
关于 buffer-memory 这个参数这里不做验证了,有兴趣的朋友可以自己去验证哈。
spring.kafka.consumer.max-poll-records 参数讨论
spring.kafka.consumer.max-poll-records 表示一次调用 poll() 操作时返回的最大记录数,默认为 500 条,上面的案例中我们使用了 properties = {“max.poll.records:10”} 这个配置,其实这个配置也是配置批量拉去消息的最大数量,我们配置的是 10,日志记录每次最多拉去的数量就是 10,使用 properties 的配置方式可以覆盖掉项目配置文件中的配置,也就是局部配置覆盖全局配置,这样做的好处是显而易见的,我们可以针对每个消费端按需做出灵活配置。
总结:本篇简单分享了 Kafka 批量发送消息消费的一些案例,希望可以帮助到有需要的朋友,分享有错误的地方也欢迎大家提出纠正。
如有不正确的地方欢迎各位指出纠正。
标签:10,27,批量,kafka,发送,消息,Kafka From: https://blog.csdn.net/weixin_42118323/article/details/143217962