集成Kafka到Spring Boot项目中的步骤和配置

简介: 集成Kafka到Spring Boot项目中的步骤和配置

集成Kafka到Spring Boot项目中的步骤和配置如下:

 

### 1. 添加依赖

 

首先,确保在`pom.xml`中添加Spring Kafka依赖:

```xml
    org.springframework.kafka
    spring-kafka
    2.7.6 
```

### 2. 配置Kafka连接信息

 

在`application.properties`或`application.yml`中配置Kafka的连接信息:

 

#### 单个Kafka Broker的配置示例:

```properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group-id
```

#### 多个Kafka Broker的配置示例:

```properties
spring.kafka.bootstrap-servers=broker1:9092,broker2:9092,broker3:9092
spring.kafka.consumer.group-id=my-group-id
```

### 3. 编写Kafka生产者

 

创建一个Kafka生产者来发送消息到指定的Topic:

```java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
 
@Component
public class KafkaProducer {
 
    @Autowired
    private KafkaTemplate kafkaTemplate;
 
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}
```

### 4. 编写Kafka消费者

 

创建一个Kafka消费者来监听指定的Topic,并处理接收到的消息:

```java
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
@Component
public class KafkaConsumer {
 
    @KafkaListener(topics = "my-topic", groupId = "my-group-id")
    public void listen(String message) {
        System.out.println("Received Message in group my-group-id: " + message);
        // 处理接收到的消息逻辑
    }
}
```

 

### 5. 配置序列化器和反序列化器(可选)

 

默认情况下,Spring Kafka使用`StringSerializer`和`StringDeserializer`来序列化和反序列化消息。如果需要使用其他格式,如JSON,可以配置对应的序列化器和反序列化器。

 

例如,配置使用JSON格式的序列化器和反序列化器:

```java
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.KafkaTemplate;
@Configuration
```

### 1. 使用JSON格式的序列化器和反序列化器

 

如果你希望在消息的传输过程中使用JSON格式,可以配置如下的序列化器和反序列化器。

 

#### 添加依赖

 

确保在`pom.xml`中添加相应的依赖:

```xml
    org.springframework.kafka
    spring-kafka
    org.springframework.kafka
    spring-kafka-json
```

#### 配置KafkaTemplate使用JSON序列化器

```java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonSerializer;
 
@Configuration
public class KafkaConfig {
 
    @Bean
    public KafkaTemplate kafkaTemplate() {
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerConfigs()),
                new JsonSerializer<>());
    }
 
    // 如果需要其他配置,比如Bootstrap地址,可以在这里进行配置
}
```

#### 配置KafkaConsumerFactory使用JSON反序列化器

 

```java
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
 
import java.util.HashMap;
import java.util.Map;
 
@Configuration
public class KafkaConfig {
 
    // 配置消费者工厂,使用JSON反序列化器
    @Bean
    public DefaultKafkaConsumerFactory kafkaConsumerFactory() {
        Map props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group-id");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); // 如果你信任所有的包,可以使用通配符*
 
        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(),
                new JsonDeserializer<>(YourMessageClass.class));
    }
 
    // 配置Kafka监听器容器
    @Bean
    public ConcurrentMessageListenerContainer kafkaListenerContainerFactory() {
        ContainerProperties containerProperties = new ContainerProperties("my-topic");
        return new ConcurrentMessageListenerContainer<>(kafkaConsumerFactory(), containerProperties);
    }
}
```

 

### 2. 手动配置Topic

 

如果需要在应用启动时自动创建Topic,可以进行如下配置:

```java
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.TopicBuilder;
 
@Configuration
@EnableKafka
public class KafkaTopicConfig {
 
    @Bean
    public NewTopic myTopic() {
        return TopicBuilder.name("my-topic")
                .partitions(1)
                .replicas(1)
                .build();
    }
}
```

### 3. 添加Kafka Admin Bean

 

如果需要在应用启动时执行Kafka管理操作(如创建Topic),可以配置KafkaAdmin Bean:

 

```java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaAdmin;
 
import java.util.HashMap;
import java.util.Map;
 
@Configuration
public class KafkaConfig {
 
    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map configs = new HashMap<>();
        configs.put("bootstrap.servers", "localhost:9092");
        return new KafkaAdmin(configs);
    }
}
```

### 总结

 

以上是一些常用的Spring Boot集成Kafka的配置和补充内容,根据你的具体需求和项目情况,可以选择性地应用这些配置。配置的核心在于正确设置Kafka的连接信息、序列化器和反序列化器,以及必要的Topic管理。

目录
相关文章
|
12天前
|
安全 Java 数据安全/隐私保护
在Java项目中集成单点登录(SSO)方案
在Java项目中集成单点登录(SSO)方案
|
11天前
|
Java 应用服务中间件 开发者
Java面试题:解释Spring Boot的优势及其自动配置原理
Java面试题:解释Spring Boot的优势及其自动配置原理
40 0
|
4天前
|
运维 DataWorks 监控
DataWorks产品使用合集之集成任务发布后,修改了任务调度的配置但没有生效,是什么导致的
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
12天前
|
SQL XML Java
解决Spring Boot项目中的数据库迁移问题
解决Spring Boot项目中的数据库迁移问题
|
13天前
|
负载均衡 Java 开发者
如何在Spring Boot项目中实现微服务架构?
如何在Spring Boot项目中实现微服务架构?
|
12天前
|
消息中间件 Java Kafka
Spring Boot与Apache Kafka Streams的集成
Spring Boot与Apache Kafka Streams的集成
|
12天前
|
Java 测试技术 Maven
在Java项目中集成单元测试与覆盖率工具
在Java项目中集成单元测试与覆盖率工具
|
13天前
|
JavaScript Java 测试技术
基于springboot+vue.js+uniapp的软件项目管理系统附带文章源码部署视频讲解等
基于springboot+vue.js+uniapp的软件项目管理系统附带文章源码部署视频讲解等
10 0
|
14天前
|
安全 Java 数据安全/隐私保护
在Java项目中集成单点登录(SSO)方案
在Java项目中集成单点登录(SSO)方案
|
14天前
|
SQL XML Java
解决Spring Boot项目中的数据库迁移问题
解决Spring Boot项目中的数据库迁移问题