Spring Boot整合Kafka教程详解

目录
  • 正文
    • 步骤一:添加依赖项
    • 步骤二:配置 Kafka
    • 步骤三:创建一个生产者
    • 步骤四:创建一个消费者

正文

本教程将介绍如何在 Spring Boot 应用程序中使用 Kafka。Kafka 是一个分布式的发布-订阅消息系统,它可以处理大量数据并提供高吞吐量。

在本教程中,我们将使用 Spring Boot 2.5.4Kafka 2.8.0

步骤一:添加依赖项

在 pom.xml 中添加以下依赖项:

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

步骤二:配置 Kafka

application.yml 文件中添加以下配置:

sping:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
    producer:
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      key-serializer: org.apache.kafka.common.serialization.StringSerializer

这里我们配置了 Kafka 的服务地址为 localhost:9092,配置了一个消费者组 ID 为 my-group,并设置了一个最早的偏移量来读取消息。在生产者方面,我们配置了消息序列化程序为 StringSerializer

步骤三:创建一个生产者

现在,我们将创建一个 Kafka 生产者,用于发送消息到 Kafka 服务器。在这里,我们将创建一个 RESTful 端点,用于接收 POST 请求并将消息发送到 Kafka。

首先,我们将创建一个 KafkaProducerConfig 类,用于配置 Kafka 生产者:

@Configuration
public class KafkaProducerConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        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> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

在上面的代码中,我们使用 @Configuration 注解将 KafkaProducerConfig 类声明为配置类。然后,我们使用 @Value 注解注入配置文件中的 bootstrap-servers 属性。

接下来,我们创建了一个 producerConfigs 方法,用于设置 Kafka 生产者的配置。在这里,我们设置了 BOOTSTRAP_SERVERS_CONFIGKEY_SERIALIZER_CLASS_CONFIGVALUE_SERIALIZER_CLASS_CONFIG 三个属性。

然后,我们创建了一个 producerFactory 方法,用于创建 Kafka 生产者工厂。在这里,我们使用了 DefaultKafkaProducerFactory 类,并传递了我们的配置。

最后,我们创建了一个 kafkaTemplate 方法,用于创建 KafkaTemplate 实例。在这里,我们使用了刚刚创建的生产者工厂作为参数,然后返回 KafkaTemplate 实例。

接下来,我们将创建一个 RESTful 端点,用于接收 POST 请求并将消息发送到 Kafka。在这里,我们将使用 @RestController 注解创建一个 RESTful 控制器:

@RestController
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @PostMapping("/send")
    public void sendMessage(@RequestBody String message) {
        kafkaTemplate.send("my-topic", message);
    }
}

在上面的代码中,我们使用 @Autowired 注解将 KafkaTemplate 实例注入到 KafkaController 类中。然后,我们创建了一个 sendMessage 方法,用于发送消息到 Kafka。

在这里,我们使用 kafkaTemplate.send 方法发送消息到 my-topic 主题。send 方法返回一个 ListenableFuture 对象,用于异步处理结果。

步骤四:创建一个消费者

现在,我们将创建一个 Kafka 消费者,用于从 Kafka 服务器接收消息。在这里,我们将创建一个消费者组,并将其配置为从 my-topic 主题读取消息。

首先,我们将创建一个 KafkaConsumerConfig 类,用于配置 Kafka 消费者:

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

在上面的代码中,我们使用 @Configuration 注解将 KafkaConsumerConfig 类声明为配置类,并使用 @EnableKafka 注解启用 Kafka。

然后,我们使用 @Value 注解注入配置文件中的 bootstrap-serversconsumer.group-id 属性。

接下来,我们创建了一个 consumerConfigs 方法,用于设置 Kafka 消费者的配置。在这里,我们设置了 BOOTSTRAP_SERVERS_CONFIG、GROUP_ID_CONFIGAUTO_OFFSET_RESET_CONFIGKEY_DESERIALIZER_CLASS_CONFIGVALUE_DESERIALIZER_CLASS_CONFIG 五个属性。

然后,我们创建了一个 consumerFactory 方法,用于创建 Kafka 消费者工厂。在这里,我们使用了 DefaultKafkaConsumerFactory 类,并传递了我们的配置。

最后,我们创建了一个 kafkaListenerContainerFactory 方法,用于创建一个 ConcurrentKafkaListenerContainerFactory 实例。在这里,我们将消费者工厂注入到 kafkaListenerContainerFactory 实例中。

接下来,我们将创建一个 Kafka 消费者类 KafkaConsumer,用于监听 my-topic 主题并接收消息:

@Service
public class KafkaConsumer {
    @KafkaListener(topics = "my-topic", groupId = "my-group-id")
    public void consume(String message) {
        System.out.println("Received message: " + message);
    }
}

在上面的代码中,我们使用 @KafkaListener 注解声明了一个消费者方法,用于接收从 my-topic 主题中读取的消息。在这里,我们将消费者组 ID 设置为 my-group-id

现在,我们已经完成了 Kafka 生产者和消费者的设置。我们可以使用 mvn spring-boot:run 命令启动应用程序,并使用 curl 命令发送 POST 请求到 http://localhost:8080/send 端点,以将消息发送到 Kafka。然后,我们可以在控制台上查看消费者接收到的消息。

这就是使用 Spring Boot 和 Kafka 的基本设置。我们可以根据需要进行更改和扩展,以满足特定的需求。

以上就是Spring Boot整合Kafka教程详解的详细内容,更多关于Spring Boot整合Kafka的资料请关注我们其它相关文章!

(0)

相关推荐

  • SpringBoot集成Kafka 配置工具类的详细代码

    目录 1.单播模式,只有一个消费者组 2.广播模式,多个消费者组 spring-kafka 是基于 java版的 kafka client与spring的集成,提供了 KafkaTemplate,封装了各种方法,方便操作,它封装了apache的kafka-client,不需要再导入client依赖 <!-- kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <arti

  • SpringBoot如何获取Kafka的Topic列表

    目录 写在前面 命令行模式 代码模式 总结 写在前面 众所周知,kafka是现代流行的消息队列,它使用经典的消息订阅发布模式实现消息的流转,大部分代码结合kafka使用都是使用它的生产者和消费者来实现消息的传递,那么对于kafka的主题的管理怎么使用代码实现呢,这是今天要讲的主题 命令行模式 kafka要结合zookeeper使用,因为它把元数据信息交给了zookeeper管理,其实使用命令行命令很容易就能对topic进行管理,主要使用的命令是kafka-topics.sh 创建主题 kafka

  • springboot+kafka中@KafkaListener动态指定多个topic问题

    目录 说明 总结一下大家问的最多的一个问题 终极方法 思路 实现 代码 总结 说明 本项目为springboot+kafak的整合项目,故其用了springboot中对kafak的消费注解@KafkaListener 首先,application.properties中配置用逗号隔开的多个topic. 方法:利用Spring的SpEl表达式,将topics 配置为:@KafkaListener(topics = “#{’${topics}’.split(’,’)}”) 运行程序,console打

  • Spring Boot整合Kafka教程详解

    目录 正文 步骤一:添加依赖项 步骤二:配置 Kafka 步骤三:创建一个生产者 步骤四:创建一个消费者 正文 本教程将介绍如何在 Spring Boot 应用程序中使用 Kafka.Kafka 是一个分布式的发布-订阅消息系统,它可以处理大量数据并提供高吞吐量. 在本教程中,我们将使用 Spring Boot 2.5.4 和 Kafka 2.8.0. 步骤一:添加依赖项 在 pom.xml 中添加以下依赖项: <dependency> <groupId>org.springfra

  • Spring Boot 集成MyBatis 教程详解

    Spring Boot是由Pivotal团队提供的全新框架,其设计目的是用来简化新Spring应用的初始搭建以及开发过程.该框架使用了特定的方式来进行配置,从而使开发人员不再需要定义样板化的配置.通过这种方式,Spring Boot致力于在蓬勃发展的快速应用开发领域(rapid application development)成为领导者. 在集成MyBatis前,我们先配置一个druid数据源. Spring Boot 系列 1.Spring Boot 入门 2.Spring Boot 属性配置

  • 在Linux系统上安装Spring boot应用的教程详解

    Unix/Linux 服务 systemd 服务 操作过程 1. 安装了JDK的centOS7虚拟机 注意下载linux版本JDK的时候不能直接通过wget这种直接链接下载,否则会解压不成功,应该打开原官网,点击同意许可后点击下载(这种方式下载很慢),比较好的方式是复制下载页的地址到迅雷,通过迅雷打开该下载页,同意许可后点击下载. 下载后解压.配置环境变量 tar -zxvf jdk1.8.0_211.jar.gz 环境变量配置:/etc/profile 文件最后添加如下 export JAVA

  • Spring Boot 整合 Reactor实例详解

    目录 引言 1 创建项目 2 集成 H2 数据库 3 创建测试类 3.1 user 实体 3.2 UserRepository 3.3 UserService 3.4 UserController 3.5 SpringReactorApplication 添加注解支持 测试 总结 引言 Reactor 是一个完全非阻塞的 JVM 响应式编程基础,有着高效的需求管理(背压的形式).它直接整合 Java8 的函数式 API,尤其是 CompletableFuture, Stream,还有 Durat

  • spring boot整合RabbitMQ实例详解(Fanout模式)

    1.Fanout Exchange介绍 Fanout Exchange 消息广播的模式,不管路由键或者是路由模式,会把消息发给绑定给它的全部队列,如果配置了routing_key会被忽略. 如上图所示,即当使用fanout交换器时,他会将消息广播到与该交换器绑定的所有队列上,这有利于你对单条消息做不同的反应. 例如存在以下场景:一个web服务要在用户完善信息时,获得积分奖励,这样你就可以创建两个对列,一个用来处理用户信息的请求,另一个对列获取这条消息是来完成积分奖励的任务. 2.代码示例 1).

  • spring boot整合CAS配置详解

    在下不才,以下是我花了好几天的时间才整合出来的在spring boot里面的CAS配置整合 为了帮助没搞定的人,毕竟自己踩了很多坑,一步一步爬过来的,有什么不足之处可以给建议  谢谢(小部分代码是整合他人的) 1.不多废话,直接上最重要的代码,以下代码整合cas的重要过程 import org.jasig.cas.client.authentication.AuthenticationFilter; import org.jasig.cas.client.session.SingleSignOu

  • Java Spring Boot消息服务万字详解分析

    目录 消息服务概述 为什么要使用消息服务 异步处理 应用解耦 流量削峰 分布式事务管理 常用消息中间件介绍 ActiveMQ RabbitMQ RocketMQ RabbitMQ消息中间件 RabbitMQ简介 RabbitMQ工作模式介绍 Work queues(工作队列模式) Public/Subscribe(发布订阅模式) Routing(路由模式) Topics(通配符模式) RPC Headers RabbitMQ安装以及整合环境搭建 安装RabbitMQ 下载RabbitMQ 安装R

  • Spring Boot Admin的使用详解(Actuator监控接口)

    第一部分 Spring Boot Admin 简介 Spring Boot Admin用来管理和监控Spring Boot应用程序. 应用程序向我们的Spring Boot Admin Client注册(通过HTTP)或使用SpringCloud®(例如Eureka,Consul)发现. UI是Spring Boot Actuator端点上的Vue.js应用程序. Spring Boot Admin 是一个管理和监控Spring Boot 应用程序的开源软件.每个应用都认为是一个客户端,通过HT

  • Spring boot 使用mysql实例详解

    Spring boot 使用mysql实例详解 开发阶段用 H2即可,上线时,通过以下配置切换到mysql,spring boot将使用这个配置覆盖默认的H2. 1.建立数据库: mysql -u root CREATE DATABASE springbootdb 2.pom.xml: <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId&g

  • Dubbo在Spring和Spring Boot中的使用详解

    一.在Spring中使用Dubbo 1.Maven依赖 <dependency> <groupId>com.alibaba</groupId> <artifactId>dubbo</artifactId> <version>2.5.3.6</version> <exclusions> <exclusion> <groupId>log4j</groupId> <artif

随机推荐