Springboot整合Kafka

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. 1. spring.kafka.bootstrap-servers:指定Kafka集群的地址列表,用于建立与Kafka集群的连接。

  2. 2. spring.kafka.consumer.group-id:定义Kafka消费者所属的组ID。消费者组ID用于将一组消费者归类为一个逻辑组,并实现负载均衡和并发消费。

  3. 3. spring.kafka.consumer.auto-offset-reset:指定消费者在启动或者发生偏移重置时的起始位置。可以设置为earliestlatestnone,分别表示从最早的可用偏移量开始消费、从最新的偏移量开始消费,或者抛出异常。

  4. 4. spring.kafka.consumer.key-deserializer:指定用于反序列化消息键的反序列化器。

  5. 5. spring.kafka.consumer.value-deserializer:指定用于反序列化消息值的反序列化器。

  6. 6. spring.kafka.producer.key-serializer:指定用于序列化消息键的序列化器。

  7. 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

欢迎大家关注我的公众号,将会为大家推荐更优质的内容!

原文始发于微信公众号(青檬小栈):Springboot整合Kafka

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

文章由极客之音整理,本文链接:https://www.bmabk.com/index.php/post/213237.html

(0)
小半的头像小半

相关推荐

发表回复

登录后才能评论
极客之音——专业性很强的中文编程技术网站,欢迎收藏到浏览器,订阅我们!