Spring Boot 整合 Kafka 基础应用

一、引言

Kafka 是一个分布式的流处理平台，具有高吞吐量、可扩展性和容错性等特点，广泛应用于日志收集、消息系统、实时数据处理等场景。Spring Boot 是一个简化 Spring 应用开发的框架，通过自动配置和约定优于配置的原则，能够快速搭建应用。本文将介绍如何使用 Spring Boot 整合 Kafka，实现基础的消息生产和消费功能。

二、环境准备

2.1 安装 Kafka

首先，需要在本地或者服务器上安装 Kafka。可以从 Kafka 官方网站（https://kafka.apache.org/downloads）下载最新版本的 Kafka。解压下载的文件后，按照官方文档的步骤启动 ZooKeeper 和 Kafka 服务。

2.2 创建 Spring Boot 项目

使用 Spring Initializr（https://start.spring.io/）创建一个新的 Spring Boot 项目，添加以下依赖：

Spring for Apache Kafka
Spring Web

在 pom.xml 中可以看到添加的依赖如下：

<dependencies>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

三、配置 Kafka

在 application.properties 或 application.yml 中配置 Kafka 的相关信息。以下是使用 application.yml 的配置示例：

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

bootstrap-servers：Kafka 服务器的地址。
producer：生产者配置，指定键和值的序列化器。
consumer：消费者配置，指定消费者组 ID、偏移量重置策略以及键和值的反序列化器。

四、创建 Kafka 生产者

创建一个 Kafka 生产者服务类，用于向 Kafka 主题发送消息。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private static final String TOPIC = "test-topic";

    public void sendMessage(String message) {
        kafkaTemplate.send(TOPIC, message);
        System.out.println("Message sent: " + message);
    }
}

在上述代码中，使用 KafkaTemplate 向名为 test-topic 的主题发送消息。

五、创建 Kafka 消费者

创建一个 Kafka 消费者服务类，用于从 Kafka 主题接收消息。

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

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

使用 @KafkaListener 注解监听 test-topic 主题，当有新消息到达时，会调用 listen 方法处理消息。

六、创建控制器

创建一个 REST 控制器，用于调用生产者发送消息。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class KafkaController {

    @Autowired
    private KafkaProducerService kafkaProducerService;

    @GetMapping("/send")
    public String sendMessage(@RequestParam("message") String message) {
        kafkaProducerService.sendMessage(message);
        return "Message sent successfully";
    }
}

在上述代码中，通过 /send 接口接收消息参数，并调用生产者服务发送消息。

七、测试应用

启动 Spring Boot 应用程序，确保 Kafka 服务正常运行。打开浏览器或使用工具（如 Postman）访问以下 URL：

http://localhost:8080/send?message=Hello,Kafka!

如果一切正常，你将在控制台看到生产者发送消息的日志和消费者接收消息的日志。

八、总结

通过以上步骤，可以成功地使用 Spring Boot 整合了 Kafka，实现了基础的消息生产和消费功能。在实际应用中，可以根据需求进一步扩展和优化，例如处理消息的序列化和反序列化、错误处理、批量生产和消费等。