当前位置：首页 > news >正文

第十一章 Stream消息驱动

news 来源：原创 2024/5/20 22:02:17

Stream消息驱动

gitee:springcloud_study: springcloud：服务集群、注册中心、配置中心（热更新）、服务网关（校验、路由、负载均衡）、分布式缓存、分布式搜索、消息队列（异步通信）、数据库集群、分布式日志、系统监控链路追踪。

1. 消息驱动概述

作用：屏蔽底层消息中间件的差异,降低切换成本，统—消息的编程模型。底层不管是什么中间件如kafka、rabbitmq，Stream可以解决不同中间件的通信。官网：Spring Cloud Stream

官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。

应用程序通过 inputs 或者 outputsj来与Spring Cloud Stream中binder对象交互。通过我们配置来binding(绑定)，而Spring Cloud Stream的 binder对象负责与消息中间件交互。所以，我们只需要搞清楚如何与Spring Cloud Stream交互就可以方便使用消息驱动的方式。

通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。
Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。

但是Stream只支持kafka、rabbitmq。

设计思想 标准的MQ：

1.生产者/消费者之间靠消息媒介传递信息内容：Message
2.消息必须走特定的通道：消息通道MessageChannel
3.消息通道里的消息如何被消费呢，谁负责收发处理：消息通道MessageChannel的子接口SubscribableChannel，由MessageHandler消息处理器所订阅

Cloud Stream：

Stream利用Binder来绑定中间件的输入流和输出流。如果系统使用到了两个中间件（kafka、rabbitmq）：这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的人—大堆东西都要重新推倒重新做，因为它跟我们的系统耦合了，这时候springcloud Stream给我们提供了一种解耦合的方式。

Stream中的消息通信方式遵循了发布-订阅模式：

Topic在Rabbitmq中是Exchange、在kafka中是Topic。

Spring Cloud Stream标准流程套路

Middleware：中间件，目前只支持RabbitMQ和Kafka
Binder：是应用与消息中间件之间的封装，目前实行了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型(对应于Kafka的topic,RabbitMQ的exchange)，这些都可以通过配置文件来实现。
@Input：注解标识输入通道，通过该输入通道接收到的消息进入应用程序
@Output：注解标识输出通道，发布的消息将通过该通道离开应用程序
@StreamListener：监听队列。用于消费者的队列的消息接收
@EnableBinding：指信道channel和exchange绑定在一起

Binder：很方便的连接中间件，屏蔽差异 Channel：通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channel对队列进行配置。 Source和Sink：简单的可理解为参照对象是Spring Cloud Stream自身，从Stream发布消息就是输出，接受消息就是输入。

2. 消息驱动之生产者

创建cloud-stream-rabbitmq-provider8801：作为生产者进行发消息模块

pom文件

<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

application.yaml

server:port: 8801
spring:application:name: cloud-stream-providercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理output: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置
eureka:client:service-url:defaultZone: http://eureka7001.com:7001/eureka,http://eureka7002.com:7002/eurekaregister-with-eureka: truefetch-registry: trueinstance:lease-renewal-interval-in-seconds: 2lease-expiration-duration-in-seconds: 5instance-id: send-8801.comprefer-ip-address: true

主启动类

@SpringBootApplication
@EnableEurekaClient
public class StreamMQMain8801 {public static void main(String[] args) {SpringApplication.run(StreamMQMain8801.class,args);}
}

service层

public interface IMessageProvider {String send();
}

@EnableBinding(Source.class)  //定义消息的推送管道
public class MessageProviderImpl implements IMessageProvider {
@Resourceprivate MessageChannel output;  //消息发送管道
@Overridepublic String send() {String serial = UUID.randomUUID().toString();output.send(MessageBuilder.withPayload(serial).build());System.out.println("********serial："+serial);return null;}
}

controller层

@RestController
public class SendMessageController {
@Resourceprivate IMessageProvider messageProvider;
@GetMapping(value = "/sendMessage")public String sendMessage(){return messageProvider.send();}
}

测试：

3. 消息驱动之消费者

创建cloud-stream-rabbitmq-consumer8802，作为消息接收模块

pom文件

<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

application.yml

server:port: 8802
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置
eureka:client:service-url:defaultZone: http://eureka7001.com:7001/eureka,http://eureka7002.com:7002/eurekaregister-with-eureka: truefetch-registry: trueinstance:lease-renewal-interval-in-seconds: 2lease-expiration-duration-in-seconds: 5instance-id: receive-8802.comprefer-ip-address: true

controller层

@Component
@EnableBinding(Sink.class)
public class ReceiveMessageController {
@Value("${server.port}")private String serverPort;
@StreamListener(Sink.INPUT)public void input(Message<String> message){System.out.println("消费者1号，------>接收到的消息："+message.getPayload()+"\t port："+serverPort);}
}

主启动类

@SpringBootApplication
@EnableEurekaClient
public class ConsumerMQMain8802 {public static void main(String[] args) {SpringApplication.run(ConsumerMQMain8802.class,args);}
}

测试：

启动loccalhost:8801/sendMessage就可以了，消费者就是一个监听器，有message就消费。

4. 分组消费与持久化

根据cloud-stream-rabbitmq-consumer8802创建8803项目，运行暴露问题：

消息重复消费和消息持久化问题，需要进行分组操作。注意在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)。

解决重复消费方法：加入同一个组（下图是不同分组的情况）

cloud-stream-rabbitmq-consumer8802和8803设置不同分组yicaiA/B

server:port: 8803
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置group: yicaiB
server:port: 8802
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置group: yicaiA

cloud-stream-rabbitmq-consumer8802和8803设置同一个组yicaiA

server:port: 8802
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置group: yicaiA

测试：