Spring Boot整合Kafka教程详解

近年来，Kafka 成为了备受关注和喜爱的消息系统，凭借其高吞吐量、低延迟的特性，成为了大型系统中消息传递和处理的首选工具。本文讲解如何在 Spring Boot 项目中整合 Kafka。

1. 添加 Maven 依赖

在 pom.xml 文件中添加以下依赖：

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

2. 配置 Kafka

在 application.yml 配置文件中添加 Kafka 的相关配置：

spring:
  kafka:
    bootstrap-servers: localhost:9092

其中，bootstrap-servers 属性指定了 Kafka 的地址和端口。

3. 发送消息到 Kafka

在 Spring Boot 中向 Kafka 发送消息，需要使用 KafkaTemplate 对象。示例代码如下：

@Service
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

该类使用 KafkaTemplate<String, String> 对象实现了消息发送操作。KafkaTemplate 对象是一个泛型类，指定了消息的 key 和 value 类型。

4. 接收来自 Kafka 的消息

在 Spring Boot 中从 Kafka 接收消息，需要使用 @KafkaListener 注解。示例代码如下：

@Service
public class KafkaConsumer {

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

@KafkaListener 注解中，topics 属性指定了要监听的主题名称，groupId 属性指定了消费者组的名称。

5. 示例说明

下面是两条示例说明，分别演示了如何向 Kafka 发送消息，以及如何从 Kafka 接收消息。

5.1 向 Kafka 发送消息

在定义了 KafkaProducer 服务后，需要在 Controller 中调用该服务，向 Kafka 发送消息。示例代码如下：

@RestController
public class MyController {

    @Autowired
    private KafkaProducer kafkaProducer;

    @PostMapping("/send-message")
    public void sendMessage(HttpServletRequest request) {
        String topic = request.getParameter("topic");
        String message = request.getParameter("message");
        kafkaProducer.sendMessage(topic, message);
    }
}

在 /send-message 接口中，获取了要发送的消息和消息主题，再调用 KafkaProducer 的 sendMessage 方法即可将消息发送到 Kafka 中。

5.2 从 Kafka 接收消息

在定义了 KafkaConsumer 服务后，需要在应用启动时通过 @EnableKafka 注解启用 Kafka 监听功能。示例代码如下：

@SpringBootApplication
@EnableKafka
public class MyApp {

    public static void main(String[] args) {
        SpringApplication.run(MyApp.class, args);
    }
}

应用启动后，KafkaConsumer 的 consume 方法将被自动调用，对消息进行处理。在控制台中会打印出接收到的消息。

至此，我们已经完成了 Spring Boot 整合 Kafka 的全部过程，可以通过以上示例代码快速上手使用 Kafka。