SpringBoot 整合 Kafka 项目实战操作详解

4.1.2 导入核心依赖

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

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

4.1.3 添加配置文件

在 application.yml 中配置 Kafka 连接信息：

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

4.2 代码整合过程

4.2.1 增加消息传递对象类

定义一个 POJO 用于封装消息内容。

public class MessageDTO {
    private String id;
    private String content;
    // getters and setters
}

4.2.2 创建发送消息的工具类

使用 KafkaTemplate 发送消息。

@Component
public class KafkaSender {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

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

4.2.3 创建消费者

使用 @KafkaListener 监听消息。

@Component
public class KafkaReceiver {
    @KafkaListener(topics = "test-topic", groupId = "group1")
    public void receive(String message) {
        System.out.println("Received: " + message);
    }
}

4.2.4 增加一个操作 topic 的工具类

使用 AdminClient 动态管理 Topic。

@Component
public class TopicManager {
    @Autowired
    private KafkaAdmin kafkaAdmin;
    // Implementation logic
}

4.2.5 创建测试接口

提供 REST API 触发消息发送。

@RestController
@RequestMapping("/api/kafka")
public class KafkaController {
    @Autowired
    private KafkaSender sender;

    @PostMapping("/send")
    public String send(@RequestParam String msg) {
        sender.send("test-topic", msg);
        return "Sent";
    }
}

4.2.6 效果验证

启动应用，调用接口，观察控制台输出。

五、写在文末

本文详细介绍了 Kafka 的基础知识及在 SpringBoot 中的集成方法。通过实践，开发者可以快速构建基于消息队列的微服务架构，提升系统性能与稳定性。