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: 1
cloud:
stream:
binders:
kafkaBiner1: # 定义一个Binder,名称随意
type: kafka # Binder的类型是 kafka
environment:
spring:
kafka: ${spring.kafka} # Binder的配置使用前面配置的Kafka的信息
default-binder: kafkaBiner1 # 默认Binder,是前面配置的Binder的名称
bindings:
kafkaChannel1: # 定义一个(作为outputs Channel)通道,名称随意,在代码中使用该通道名称即可
binder: kafkaBiner1 # 使用kafkaBiner1
destination: KafkaFirstTopic # 定义目标Topic的名称
kafkaChannel2: # 定义一个(作为inputs Channel)通道,名称随意,在代码中使用该通道名称即可
binder: kafkaBiner1 # 使用kafkaBiner1
destination: KafkaFirstTopic # 定义目标Topic的名称
group: group0 # 作为消息的消费方,需要指定group
集成生产者
下面开发一个生产者,发送消息需要通过outputs Channel进行,使用kafkaChannel1
发送消息到Kafka。
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@Api("生产者接口")
@RestController
@RequestMapping("/producer")
public class ProducerController {
@Autowired
private StreamBridge bridge;
@ApiOperation("向Kafka发送数据")
@PostMapping("/kafka")
public String sendToKafka(String message) {
boolean status = bridge.send("kafkaChannel1", message);
return "发送消息:" + message + "=====>" + status;
}
}
集成消费者
消费消息需要通过inputs Channel进行,定义一个Processor,指定订阅通道为kafkaChannel2
,这个通道被用于进行消息消费,需要定义group。
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;
@Component
public interface ConsumerProcessor {
@Input("kafkaChannel2")
SubscribableChannel subscribableChannel();
}
启用通道并监听。
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
@EnableBinding(ConsumerProcessor.class)
public class ConsumerProcessorImpl {
@StreamListener("kafkaChannel2")
public void kafkaStreamListener(Object message) {
System.out.println("接收到Kafka消息:" + new String((byte[]) message));
}
}
集成验证
生产者验证
首先启动一个Kafka自带的消费者,监听KafkaFirstTopic
。
接下来启动SpringBoot项目并发送消息。在消费者那里可以看到接收到的消息。
消费者验证
前面消息已经发送到了Kafka的Topic了,可以看到控制台直接打印出了监听到的消息。文章来源:https://www.toymoban.com/news/detail-643671.html
至此,Spring Cloud Stream集成Kafka完成。文章来源地址https://www.toymoban.com/news/detail-643671.html
到了这里,关于Spring Cloud Stream集成Kafka的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!