当前位置：首页 > news >正文

Spring Boot集成kafka的相关配置

news 2025/7/13 7:57:35

引入依赖：

额外依赖只需要这一个，kafka-client 不是springboot 的东西，那是原生的 kafka 客户端， kafka-test也不需要，是用代码控制broker的东西。

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

yml配置：

也可以用java类Config 方式配置，如果没有特殊要求，可以只用spring配置的方式

server:port: 8080
spring:kafka:# Kafka服务器，支持集群bootstrap-servers: 127.0.0.1:9092,127.0.0.2:9092# 生产者配置producer:# 消息发送重试次数，注意会引起重复消费，消费者需要做幂等retries: 3# acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。# acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。# acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。# 开启事务时，必须设置为allacks: all# 当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。batch-size: 16384# 生产者内存缓冲区的大小。buffer-memory: 1024000# 键的序列化方式key-serializer: org.apache.kafka.common.serialization.StringSerializer# 值的序列化方式value-serializer: org.apache.kafka.common.serialization.StringSerializer# 消费者配置consumer:# 消费组ID，同一个消费组不会重复消费数据group-id: testGroup# 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D#auto-commit-interval: 1S# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：# earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费分区的记录# latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据（在消费者启动之后生成的记录）# none：当各分区都存在已提交的offset时，从提交的offset开始消费；只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: latest# 是否自动提交偏移量，默认值是true，为了避免出现重复数据和数据丢失，可以把它设置为false，然后手动提交偏移量enable-auto-commit: false# 键的序列化方式key-serializer: org.apache.kafka.common.serialization.StringSerializer# 值的序列化方式value-serializer: org.apache.kafka.common.serialization.StringSerializer# 这个参数定义了poll方法最多可以拉取多少条消息，默认值为500。如果在拉取消息的时候新消息不足500条，那有多少返回多少；如果超过500条，每次只返回500。# 这个默认值在有些场景下太大，有些场景很难保证能够在5min内处理完500条消息，# 如果消费者无法在5分钟内处理完500条消息的话就会触发reBalance,# 然后这批消息会被分配到另一个消费者中，还是会处理不完，这样这批消息就永远也处理不完。# 要避免出现上述问题，提前评估好处理一条消息最长需要多少时间，然后覆盖默认的max.poll.records参数# 注：需要开启BatchListener批量监听才会生效，如果不开启BatchListener则不会出现reBalance情况max-poll-records: 3properties:# 两次poll之间的最大间隔，默认值为5分钟。如果超过这个间隔会触发reBalancemax:poll:interval:ms: 600000# 当broker多久没有收到consumer的心跳请求后就触发reBalance，默认值是10ssession:timeout:ms: 10000listener:# 在侦听器容器中运行的线程数，一般设置为 机器数*分区数concurrency: 4# 自动提交关闭，需要设置手动消息确认ack-mode: manual_immediate# 消费监听接口监听的主题不存在时，默认会报错，所以设置为false忽略错误missing-topics-fatal: false# 两次poll之间的最大间隔，默认值为5分钟。如果超过这个间隔会触发reBalancepoll-timeout: 600000#批量提交#      type: batch

业务代码：

简单生产

@RestController
public class kafkaProducer {@Autowiredprivate KafkaTemplate<String, Object> kafkaTemplate;@GetMapping("/kafka/normal/{message}")public void sendNormalMessage(@PathVariable("message") String message) {kafkaTemplate.send("testTopic", message);}
}

简单消费

注意加上@Component，被spring管理监听才有效

@Component
public class KafkaConsumer {//监听消费@KafkaListener(topics = {"testTopic"})public void onNormalMessage(ConsumerRecord<String, Object> record) {System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +record.value());}}

主题使用配置的方式：

在yml配置里加上topic

server:port: 8080
spring:kafka:# Kafka服务器，支持集群bootstrap-servers: 127.0.0.1:9092,127.0.0.2:9092# 自定义配置myconfig:topic: testTopic

简单生产+配置主题

@RestController
public class kafkaProducer {@Autowiredprivate KafkaTemplate<String, Object> kafkaTemplate;@Value("${spring.kafka.myconfig.topic}")private String topic;@GetMapping("/kafka/normal/{message}")public void sendNormalMessage(@PathVariable("message") String message) {kafkaTemplate.send(topic, message);}
}

简单消费+配置主题

注意这里不能用@Value注解的方式，会报错

@Component
public class KafkaConsumer {//监听消费@KafkaListener(topics = {"${spring.kafka.myconfig.topic}"})public void onNormalMessage(ConsumerRecord<String, Object> record) {System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +record.value());}}

如果是多个主题需要消费，可以使用Spring的SpEl表达式

server:port: 8080
spring:kafka:# Kafka服务器，支持集群bootstrap-servers: 127.0.0.1:9092,127.0.0.2:9092# 自定义配置myconfig:topic: testTopic1,testTopic2

@KafkaListener(topics = {"#{'${spring.kafka.myconfig.topic}'.split(',')}"})

@Component
public class KafkaConsumer {//监听消费@KafkaListener(topics = {"#{'${spring.kafka.myconfig.topic}'.split(',')}"})public void onNormalMessage(ConsumerRecord<String, Object> record) {System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +record.value());}}

Kafka数据重复和数据丢失的解决方案_kafka相同数据_谢小涛的博客-CSDN博客

查看全文

http://www.lryc.cn/news/191677.html