Spring Cloud Stream集成Kafka

Spring Cloud Stream集成Kafka

Spring Cloud Stream是一个构建消息驱动微服务的框架，抽象了MQ的使用方式, 提供统一的API操作。Spring Cloud Stream通过Binder(绑定器)、inputs/outputs Channel完成应用程序和MQ的解耦。

• Binder 负责绑定应用程序和MQ中间件,即指定应用程序是和KafKa交互还是和RabbitMQ交互或者和其他的MQ中间件交互
• inputs/outputs Channel inputs/outputs Channel抽象发布订阅消息的方式，即无论是什么类型的MQ应用程序都通过统一的方式发布订阅消息

我们已经搭建好了Kafka（参考Kafka单节点安装），本文主要介绍一下Spring Cloud Stream与Kafka进行集成实现消息的生产及消费。

项目创建

首先需要创建一个SpringBoot项目，命名为：

spring-integration-kafka

，在配置文件中导入相关的依赖。
项目情况为：

• 构建工具：Gradle
• SpringBoot版本：2.7.5
• SpringBoot依赖管理版本：1.0.15.RELEASE
• SpringCloud依赖管理版本：2021.0.5

项目依赖

配置文件build.gradle.kts的关键配置项如下：

plugins {
    id("org.springframework.boot") version "2.7.5"
    id("io.spring.dependency-management") version "1.0.15.RELEASE"
}

apply(plugin = "org.springframework.boot")
apply(plugin = "io.spring.dependency-management")
apply(plugin = "java")

extra["springCloudVersion"] = "2021.0.5"

dependencyManagement {
    imports {
        mavenBom("org.springframework.cloud:spring-cloud-dependencies:${property("springCloudVersion")}")
    }
}

dependencies {
    implementation("org.springframework.boot:spring-boot-starter-web")
    implementation("org.springframework.boot:spring-boot-starter-actuator")
    
    implementation("org.springframework.cloud:spring-cloud-starter-bootstrap")
    implementation("org.springframework.cloud:spring-cloud-starter-stream-kafka")
    
    implementation("io.springfox:springfox-boot-starter:3.0.0")
    implementation("com.github.xiaoymin:swagger-bootstrap-ui:1.9.6")
}

集成配置

定义配置文件

application.yml

，配置文件中主要配置Kafka的地址、以及Spring Colud Stream的Binder和inputs/outputs Channel，其中：kafkaChannel1用于向Kafka发送消息；kafkaChannel2用于消费Kafka的消息。

spring:kafka:bootstrap-servers: wux-labs-vm:9092# 定义Kafka的地址producer:acks:1cloud:stream:binders:kafkaBiner1:# 定义一个Binder，名称随意type: kafka   # Binder的类型是 kafkaenvironment:spring:kafka: ${spring.kafka}# Binder的配置使用前面配置的Kafka的信息default-binder: kafkaBiner1    # 默认Binder，是前面配置的Binder的名称bindings:kafkaChannel1:# 定义一个（作为outputs Channel）通道，名称随意，在代码中使用该通道名称即可binder: kafkaBiner1 # 使用kafkaBiner1destination: KafkaFirstTopic # 定义目标Topic的名称kafkaChannel2:# 定义一个（作为inputs Channel）通道，名称随意，在代码中使用该通道名称即可binder: kafkaBiner1 # 使用kafkaBiner1destination: KafkaFirstTopic # 定义目标Topic的名称group: group0   # 作为消息的消费方，需要指定group

集成生产者

下面开发一个生产者，发送消息需要通过outputs Channel进行，使用

kafkaChannel1

发送消息到Kafka。

importio.swagger.annotations.Api;importio.swagger.annotations.ApiOperation;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.cloud.stream.function.StreamBridge;importorg.springframework.web.bind.annotation.PostMapping;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RestController;@Api("生产者接口")@RestController@RequestMapping("/producer")publicclassProducerController{@AutowiredprivateStreamBridge bridge;@ApiOperation("向Kafka发送数据")@PostMapping("/kafka")publicStringsendToKafka(String message){boolean status = bridge.send("kafkaChannel1", message);return"发送消息："+ message +"=====>"+ status;}}

集成消费者

消费消息需要通过inputs Channel进行，定义一个Processor，指定订阅通道为

kafkaChannel2

，这个通道被用于进行消息消费，需要定义group。

importorg.springframework.cloud.stream.annotation.Input;importorg.springframework.messaging.SubscribableChannel;importorg.springframework.stereotype.Component;@ComponentpublicinterfaceConsumerProcessor{@Input("kafkaChannel2")SubscribableChannelsubscribableChannel();}

启用通道并监听。

importorg.springframework.cloud.stream.annotation.EnableBinding;importorg.springframework.cloud.stream.annotation.StreamListener;@EnableBinding(ConsumerProcessor.class)publicclassConsumerProcessorImpl{@StreamListener("kafkaChannel2")publicvoidkafkaStreamListener(Object message){System.out.println("接收到Kafka消息："+newString((byte[]) message));}}

集成验证

生产者验证

首先启动一个Kafka自带的消费者，监听

KafkaFirstTopic

。

接下来启动SpringBoot项目并发送消息。在消费者那里可以看到接收到的消息。

消费者验证

前面消息已经发送到了Kafka的Topic了，可以看到控制台直接打印出了监听到的消息。

至此，Spring Cloud Stream集成Kafka完成。