Spring Boot集成Apache Kafka的实战指南

目录

• 开发环境准备
• 创建 Spring Boot 项目
• 配置 Kafka 连接信息
• 编写 Kafka 生产者（Producer）
• 编写 Kafka 消费者（Consumer）
• 添加 REST 接口用于测试发送消息
• 启动 Kafka 环境（可选）
• 启动 Zookeeper（Kafka 依赖）
• 启动 Kafka 服务
• 创建测试 Topic
• 测试接口
• 扩展功能建议

Apache Kafka 是一个分布式流处理平台，广泛用于构建实时数据管道、日志聚合系统和事件溯源架构。Spring Boot 提供了对 Kafka 的良好集成支持，使得开发者可以非常便捷地在项目中使用 Kafka。

本文将手把手教你如何在 Spring Boot 项目中集成 Kafka，包括生产者（Producer）和消费者（Consumer）的实现，并提供完整的代码示例。

开发环境准备

Java 17+

Maven 或 Gradle

Spring Boot 3.x

Apache Kafka 3.0+（本地或远程）

IDE（如 IntelliJ IDEA、VS Code）

创建 Spring Boot 项目

你可以通过 Spring Initializr 创建一个新的 Spring Boot 项目，选择以下依赖：

• Spring Web
• Spring for Apache Kafka

或者手动添加 pom.XML 中的依赖：

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

Spring Boot 会自动管理版本兼容性，无需手动指定版本号。

配置 Kafka 连接信息

在 application.yml 或 application.properties 文件中配置 Kafka 相关参数：

application.yml 示例：

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

编写 Kafka 生产者（Producer）

创建一个服务类用于发送消息到 Kafka 主题。

KafkaProducer.java

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

@Service
public class KafkaProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    p编程ublic KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

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

编写 Kafka 消费者（Consumer）

使用 @KafkaListener 注解监听特定主题的消息。

KafkaConsumer.java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumer {
android
    @KafkaListener(topics = "test-topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.printf("Received message: topic - %s, partition - %d, offset - %d, key - %s, value - %s%n",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
}
http://www.devze.com

添加 REST 接口用于测试发送消息

为了方便测试，我们可以创建一个简单的 REST 控制器来触发消息发送。

KafkaController.java

import org.springframework.web.bind.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;

@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private KafkaProducer kafkaProducer;

    @PostMapping("/send")
    public String sendMeslOIPlsBssage(@RequestParam String msg) {
        kafkaProducer.sendMessage("test-topic", msg);
        return "Message sent: " + msg;
    }
}

启动 Kafka 环境（可选）

如果你还没有运行 Kafka，可以按照以下步骤快速启动：

启动 Zookeeper（Kafka 依赖）

bin/zookeeper-server-start.sh config/zookeeper.properties

启动 Kafka 服务

bin/kafka-server-start.sh config/server.properties

创建测试 Topic

bin/kafka-topics.sh --create --topic test-topic --bootstrap-server lopythoncalhost:9092 --partitions 1 --replication-factor 1

测试接口

启动 Spring Boot 应用后，访问如下接口发送消息：

POST http://localhost:8080/kafka/send?msg=HelloKafka

观察控制台输出，确认是否收到类似以下内容：

Received message: topic - test-topic, partition - 0, offset - 5, key - null, value - HelloKafka

扩展功能建议

使用 jsON 格式传输对象（自定义序列化/反序列化）

多消费者组配置与负载均衡

异常处理与重试机制（@DltHandler, SeekToCurrentErrorHandler）

Kafka Streams 实现实时流处理逻辑

配置 SSL、SASL 安全认证

结合 Spring Cloud Stream 构建云原生事件驱动架构

到此这篇关于Spring Boot集成Apache Kafka的实战指南的文章就介绍到这了,更多相关SpringBoot集成Apache Kafka内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！