Springboot 集成kafka

### Spring Boot 整合 Kafka 示例教程 #### 创建 Spring Boot 项目并添加依赖项 为了开始整合过程，在构建工具中引入必要的依赖项。对于 Maven 用户，可以在 `pom.xml` 文件内加入如下片段： ```xml <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> ``` 这一步骤确保了应用程序能够访问到操作 Apache Kafka 所需的功能库[^2]。 #### 配置 Kafka 连接参数 编辑项目的配置文件 `application.properties` 或者 `application.yml` 来指定连接至 Kafka Broker 的地址以及其他必要选项。例如： ```properties ccm.kafka.servers=192.168.1.95:9092,192.168.1.96:9092,192.168.1.97:9092 ccm.kafka.topics.xxx=xxx_content_dev ``` 上述设置指定了多个 Kafka 节点作为集群成员，并定义了一个用于测试目的的主题名称[^4]。 #### 编写生产者组件 接下来开发一个简单的消息生产者类来向指定主题发布数据包。这里给出了一种可能的设计模式——利用 `@KafkaListener` 注解监听特定话题上的事件流；而发送端则借助于 `KafkaTemplate` 实现异步投递机制。 ```java @Service public class MyMessageProducer { private final KafkaTemplate<String, String> kafkaTemplate; @Autowired public MyMessageProducer(KafkaTemplate<String, String> kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; } public void sendMessage(String topicName, String message){ ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName,message); try{ SendResult<String,String> result=future.get(); System.out.println("Sent message=[" +message+ "] with offset=["+result.getRecordMetadata().offset()+"]"); }catch(InterruptedException | ExecutionException e){ throw new RuntimeException(e.getMessage()); } } } ``` 这段代码展示了怎样封装一条记录并通过网络传输给目标节点群组[^1]。 #### 构建消费者逻辑 同样地，也需要准备相应的接收方程序段落处理传入的信息单元。下面的例子说明了如何声明一个批处理器实例以及开启自动启动功能以便即时响应新到达的通知事项。 ```java @Component @EnableKafka public class BatchConsumer { @Value("${kafka.topic.name}") private String topicName; @KafkaListener(topics = "${kafka.topic.name}", groupId = "group_id", containerFactory="batchContainerFactory") public void listen(List<ConsumerRecord<?, ?>> records) { for (var record : records) { System.out.printf("key = %s, value = %s%n", record.key(), record.value()); } } static { System.setProperty("autoStartup","true"); } } ``` 注意这里的静态初始化器用来激活默认行为开关，使得整个监听进程能够在应用加载完成后立即生效[^3]。 ---