前言：
上一篇分享了 Kafka 的一些基本概念及应用场景，本篇我们来分享一下在 Spring Boot 项目中如何使用 Kafka。
Kafka 系列文章传送门
Kafka 简介及核心概念讲解
Spring Boot 集成 Kafka
引入 Kafka 依赖
在项目的 pom.xml 文件中引入 Kafka 依赖，如下：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
<version>3.1.6</version>
</dependency>
添加 Kafka 配置
在 application.properties 文件中配置 Kafka 相关配置，如下：
#kafka server 地址
spring.kafka.bootstrap-servers=10.xxx.4.xxx:9092,10.xxx.4.xxx::9092,10.xxx.4.xxx::9092
spring.kafka.producer.acks = 1
spring.kafka.producer.retries = 0
spring.kafka.producer.batch-size = 30720000
spring.kafka.producer.buffer-memory = 33554432
spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializer
#消费者配置
spring.kafka.consumer.group-id = test-kafka
#是否开启手动提交 默认自动提交
spring.kafka.consumer.enable-auto-commit = true
#如果enable.auto.commit为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000 自动提交已消费offset时间间隔
spring.kafka.consumer.auto-commit-interval = 5000
#earliest:分区已经有提交的offset从提交的offset开始消费,如果没有提交的offset,从头开始消费,latest:分区下已有提交的offset从提交的offset开始消费,没有提交的offset从新产生的数据开始消费
spring.kafka.consumer.auto-offset-reset = earliest
spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.StringDeserializer
#kafka 没有创建指定的 topic 下 项目启动是否报错 true false
spring.kafka.listener.missing-topics-fatal = false
#Kafka 的消费模式 single 每次单条消费消息 batch 每次批量消费消息
spring.kafka.listener.type = single
#一次调用 poll() 操作时返回的最大记录数 默认为 500 条
spring.kafka.consumer.max-poll-records = 2
#消息 ACK 模式 有7种
spring.kafka.listener.ack-mode = manual_immediate
#kafka session timeout
spring.kafka.consumer.session.timeout.ms = 300000
配置 Kafka Producer
我们创建一个配置类，并配置生产者工厂，配置 KafkaTemplate。
package com.order.service.config;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import java.util.HashMap;
import java.util.Map;
/**
* @author ：author
* @description：
* @modified By：
* @version: V1.0
*/
@Configuration
@EnableKafka
public class KafkaProducerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String servers;
@Bean("myProducerKafkaProps")
public Map<String, Object> getMyKafkaProps() {
Map<String, Object> props = new HashMap<>(4);
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
@Bean
public ProducerFactory<String, String> newProducerFactory() {
return new DefaultKafkaProducerFactory<>(getMyKafkaProps());
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(newProducerFactory());
}
}
配置 Kafka Cousumer
我们创建一个配置类，配置消费者工厂和监听容器。
package com.order.service.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import java.util.HashMap;
import java.util.Map;
/**
* @author ：author
* @description：
* @modified By：
* @version: V1.0
*/
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String servers;
@Value("${spring.kafka.consumer.group-id}")
private String groupId;
@Value("${spring.kafka.consumer.auto-offset-reset}")
private String offsetReset;
@Value("${spring.kafka.consumer.max-poll-records}")
private String maxPollRecords;
@Value("${spring.kafka.consumer.auto-commit-interval}")
private String autoCommitIntervalMs;
@Value("${spring.kafka.consumer.enable-auto-commit}")
private boolean enableAutoCommit;
@Bean("myConsumerKafkaProps")
public Map<String, Object> getMyKafkaProps() {
Map<String, Object> props = new HashMap<>(12);
//是否自动提交
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
//kafak 服务器
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//不存在已经提交的offest时 earliest 表示从头开始消费,latest 表示从最新的数据消费
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetReset);
//消费组id
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
//一次调用poll()操作时返回的最大记录数，默认值为500
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
//自动提交时间间隔 默认 5秒
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs);
//props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
return props;
}
/**
* @return org.springframework.kafka.config.KafkaListenerContainerFactory<org.springframework.kafka.listener.ConcurrentMessageListenerContainer < java.lang.String, java.lang.String>>
* @date 2024/10/22 19:41
* @description kafka 消费者工厂
*/
@Bean("myContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(getMyKafkaProps()));
// 并发创建的消费者数量
factory.setConcurrency(3);
// 开启批处理
factory.setBatchListener(true);
//拉取超时时间
factory.getContainerProperties().setPollTimeout(1500);
//是否自动提交 ACK kafka 默认是自动提交
if (!enableAutoCommit) {
//共有其中方式
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
}
return factory;
}
}
Kafka 消息发送
创建一个 Kafka 的 Producer，注入 KafkaTemplate，完成消息发送。
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;
/**
* @ClassName： KafkaProducer
* @Author: Author
* @Date: 2024/10/22 19:22
* @Description:
*/
@Slf4j
@Component
public class MyKafkaProducer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage( String message) {
this.kafkaTemplate.send("my-topic", message);
}
}
Kafka 消息消费
创建一个 Kafka 的 Consumer，使用 @KafkaListener 注解完成消息消费。
package com.order.service.kafka.consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
* @ClassName： KafkaConsumer
* @Author: Author
* @Date: 2024/10/22 19:22
* @Description:
*/
@Slf4j
@Component
public class MyKafkaConsumer {
@KafkaListener(id = "my-kafka-consumer",
idIsGroup = false, topics = "my-topic",
containerFactory = "myContainerFactory")
public void listen(String message) {
log.info("消息消费成功,消息内容:{}", message);
}
}
Kafka 消息发送消费测试
触发消息发送后，得到如下结果：
2024-10-22 20:22:43.041 INFO 36496 --- [-consumer-0-C-1] c.o.s.kafka.consumer.MyKafkaConsumer : 消息消费成功,消息内容:第一条 kafka 消息
结果符合预期。
以上我们简单的分享了使用 Spring Boot 集成 Kafka 的过程，希望帮助到有需要的朋友。
如有不正确的地方欢迎各位指出纠正。
