springboot集成Kafka集群

### Spring Boot 整合 Kafka 集群教程 #### 1. 创建 Spring Boot 项目并添加依赖 为了使 Spring Boot 应用程序能够与 Kafka 进行交互，在 `pom.xml` 文件中需引入必要的 Maven 依赖项。具体来说，应包含用于简化消息传递应用程序开发的 Spring Cloud Stream Kafka Starter。 ```xml <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-kafka</artifactId> <version>3.1.6</version> </dependency> ``` 此依赖不仅提供了访问 Apache Kafka 所必需的基础组件，还集成了自动配置功能来加速项目的搭建过程[^3]。 #### 2. 启动 Zookeeper 和 Kafka 服务 在本地环境中运行 Kafka 前，先要确保已经正确安装并启动了 Zookeeper 及其关联的服务实例。这一步骤对于构建可靠的分布式事件流平台至关重要。可以通过 Docker 来快速部署这些基础设施组件，从而减少环境设置的时间成本和复杂度。 #### 3. 配置 application.yml 文件 接下来调整应用的主要配置文件 `application.yml` 或者 `.properties` 版本，以便指定连接至目标 Kafka 实例所需的参数： ```yaml spring: cloud: stream: bindings: inputChannel: # 定义输入通道名称 destination: my-topic-name # 设置主题名作为目的地 group: consumer-group-id # 设定消费者组ID outputChannel: # 定义输出通道名称 destination: another-topic-name # 指向另一个不同或相同的主题 kafka: binder: brokers: localhost:9092 # 列表形式提供Kafka broker地址列表 ``` 上述 YAML 结构化数据片段展示了如何通过绑定机制将内部逻辑单元映射到外部资源上，并指定了通信端点以及参与者的身份标识信息。 #### 4. 编写生产者代码 利用 `@EnableBinding(Source.class)` 注解激活消息源接口的支持后，就可以方便地发送消息给指定的目标位置了。下面是一个简单的例子展示怎样创建一个 RESTful Web Service 接口用来触发消息发布行为： ```java @RestController public class MessageController { @Autowired private Source source; @PostMapping("/send") public void sendMessage(@RequestParam String message){ this.source.output().send(MessageBuilder.withPayload(message).build()); } } ``` 这段 Java 代码定义了一个控制器类接收 HTTP POST 请求并将接收到的内容封装成一条新消息推送到预设好的队列里去[^1]。 #### 5. 编写消费者代码 同样借助于 `@StreamListener` 注解监听特定的消息通道上的活动情况，一旦检测到来自上游节点的数据包就会立即执行相应的处理函数。这里给出一段示范性的 Consumer 组件实现方式： ```java @Component @EnableBinding(Sink.class) public class MyConsumer { @StreamListener(target = Sink.INPUT, condition="headers['type']=='greeting'") public void receiveMessage(String payload) { System.out.println("Received greeting message: " + payload); } @StreamListener(target = Sink.INPUT, condition="headers['type']!='greeting'") public void handleOtherMessages(String payload) { System.out.println("Handling non-greeting message: " + payload); } } ``` 以上两个方法分别针对不同类型的消息设置了不同的响应策略，实现了基于内容过滤的选择性消费模式。