Spring Boot提供了简化的配置和自动化的任务,使得与Kafka的集成变得更加容易。它提供了许多现成的库和工具,可以快速搭建和配置Kafka生产者和消费者。
所需依赖
本文所使用的是SpringBoot2.5.5
版本,下面是maven项目的pom.xml
文件所需的依赖:
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- Spring Kafka -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.7.1</version>
</dependency>
<!-- Kafka Clients -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.0.0</version>
</dependency>
</dependencies>
添加配置
编辑application.yml
文件:
spring:
kafka:
# 这里改成你的kafka的主机IP
# 可以参考一起推送的安装教程
bootstrap-servers: 192.168.5.200:9092
consumer:
group-id: kafka-group
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
配置解析
-
1.
spring.kafka.bootstrap-servers
:指定Kafka集群的地址列表,用于建立与Kafka集群的连接。 -
2.
spring.kafka.consumer.group-id
:定义Kafka消费者所属的组ID。消费者组ID用于将一组消费者归类为一个逻辑组,并实现负载均衡和并发消费。 -
3.
spring.kafka.consumer.auto-offset-reset
:指定消费者在启动或者发生偏移重置时的起始位置。可以设置为earliest
、latest
或none
,分别表示从最早的可用偏移量开始消费、从最新的偏移量开始消费,或者抛出异常。 -
4.
spring.kafka.consumer.key-deserializer
:指定用于反序列化消息键的反序列化器。 -
5.
spring.kafka.consumer.value-deserializer
:指定用于反序列化消息值的反序列化器。 -
6.
spring.kafka.producer.key-serializer
:指定用于序列化消息键的序列化器。 -
7.
spring.kafka.producer.value-serializer
:指定用于序列化消息值的序列化器。
创建生产者
@Component
public class KafkaProducer {
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage(String topic, String message) {
kafkaTemplate.send(topic, message);
}
}
创建消费者
@Component
public class KafkaConsumer {
@KafkaListener(topics = "test-topic", groupId = "kafka-group")
public void consumeMessage(String message) {
log.info("Received Message: {}", message);
}
}
启动测试
在启动类中进行简单的测试:
@SpringBootApplication
public class SpringbootKafkaApplication implements CommandLineRunner {
@Resource
private KafkaProducer kafkaProducer;
public static void main(String[] args) {
SpringApplication.run(SpringbootKafkaApplication.class, args);
}
@Override
public void run(String... args) throws Exception {
kafkaProducer.sendMessage("test-topic", "Hello, Kafka!");
}
}
启动成功后可以看到日志输出内容:
欢迎大家关注我的公众号,将会为大家推荐更优质的内容!
原文始发于微信公众号(青檬小栈):Springboot整合Kafka
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
文章由极客之音整理,本文链接:https://www.bmabk.com/index.php/post/213237.html