SpringCloud Stream消息驱动
消息驱动概述是什么什么是SpringCloudStream
官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。
应用程序通过inputs或者outputs与Spring Cloud Stream中binder对象交互。
通过我们配置来binding(绑定),而Spring Cloud Stream的binder对象负责与消息中间件交互。
所以,我们只需要搞清楚如何与Spring Cloud Stream父互就可以方便使用消息驱动的方式。
通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。
Spring Cloud Stream为-些供应商的消息 中间件产品提供了个性化的自动化配置实现,引用了发布订阅、消费组、分区的三个核心概念。
目前仅支持RabbitMQ、Kafka。
一句话屏蔽底层消息中间件的差异,降低切换成本,统一消息的编程模型
官网https://spring.io/projects/spring-cloud-stream
Spring Cloud Stream中文指导手册
https://blog.csdn.net/qq_32734365/article/details/81413218#spring-cloud-stream%E4%B8%AD%E6%96%87%E6%8C%87%E5%AF%BC%E6%89%8B%E5%86%8C
Message
消息必须走特定的通道消息通道MessageChannel
消息通道里的消息如何被消费呢,谁负责收发处理消息通道MessageChannel的子接口SubscribableChannel,由MessageHandler消息处理器所订阅
为什么使用Cloud Stream比方说我们用到了RabbitMQ和Kafka,于这两个消息中间件的架构上的不同,
像RabbitMQ有exchange,kafka有 topic和Partitions分区,
在没有绑定器这个概念的情况下,我们的SpringBoot应要直接与消息中间件进行信息交互的时候,
由于各消息中间件构建的初衷不同,它们的实现细节上会有较大的差异性
通过定义绑定器作为中间层,完美地实现了应用程序与消息中间件细节之间的隔离。
通过向应用程序暴露统一的Channel通道,使得应用程序不需要再考虑各种不同的消息中间件实现。
通过定义绑定器Binder作为中间层,实现了应用程序与消息中间件细节之间的隔离。
Binder在没有绑定器这个概念的情况下,我们的SpringBoot应用要 直接与消息中间件进行信息交互的时候,由于各消息中间件构建的初衷不同,它们的实现细节上会有较大的差异性.通过定义绑定器作为中间层,完美地实现了应用程序与消息中间件细节之间的隔离。Stream对消 息中间件的进一步封装可以做到代码层面对中间件的无感知,甚于动态的切换中间件(rabbitmq切换为kafka),使得微服务开发的高度解耦,服务可以关注更多自己的业务流程。
很方便的连接中间件,屏蔽差异
Channel通道,是队列Queue的一种抽象,在消息通讯系统中就是实现存储和转发的媒介,通过Channel对队列进行配置
Source和Sink简单的可以理解为参照对象是Spring Cloud Stream 自身,从Stream发布消息就是输出,接受消息就是输入
编码API和常用注解案例说明RabbitMQ环境已经OK工程中新建三个子模块cloud-stream-rabbitmq-provIDer8801 ,作为消息试生产者进行发消息模块cloud-stream-rabbitmq-consumer8802,作为消息接收模块cloud-stream-rabbitmq-consumer8803,作为消息接收模块消息驱动之生产者新建Modulecloud-stream-rabbitmq-provIDer8801
POM<dependencIEs> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-starter-web</artifactID> </dependency> <!--监控--> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-starter-actuator</artifactID> </dependency> <!--eureka clIEnt--> <dependency> <groupID>org.springframework.cloud</groupID> <artifactID>spring-cloud-starter-netflix-eureka-clIEnt</artifactID> </dependency> <!--stream rabbit --> <dependency> <groupID>org.springframework.cloud</groupID> <artifactID>spring-cloud-starter-stream-rabbit</artifactID> </dependency> <!--热部署--> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-devtools</artifactID> <scope>runtime</scope> <optional>true</optional> </dependency> <dependency> <groupID>org.projectlombok</groupID> <artifactID>lombok</artifactID> <optional>true</optional> </dependency> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-starter-test</artifactID> <scope>test</scope> </dependency></dependencIEs>
YMLserver: port: 8801spring: application: name: cloud-stream-provIDer cloud: stream: binders: #在此处配置要绑定的rabbitmq的服务信息 defaultRabbit: #表示定义的名称,用于binding整合 type: rabbit #消息组件类型 environment: #设置rabbitmq的相关环境配置 spring: rabbitmq: host: localhost port: 5672 username: guest password: guest bindings: #服务的整合处理 output: #这个名字是一个通道的名称 destination: studyExchange #表示要使用的Exchange名称定义 content-type: application/Json #设置消息类型,本次为Json,本文要设置为“text/plain” binder: defaultRabbit #设置要绑定的消息服务的具体设置eureka: clIEnt: service-url: defaultZone: http://localhost:7001/eureka instance: lease-@R_475_5026@-interval-in-seconds: 2 #设置心跳的时间间隔(默认是30S) lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s instance-ID: send-8801.com #在信息列表时显示主机名称 prefer-ip-address: true #访问的路径变为IP地址
主启动类StreamMQMain8801@SpringBootApplicationpublic class StreamMQMain8801 { public static voID main(String[] args) { SpringApplication.run(StreamMQMain8801.class,args); }}
业务类发送消息接口public interface IMessageProvIDer { public String send();}
发送消息接口实现类package com.eiletxIE.springcloud.service.impl;import com.eiletxIE.springcloud.service.IMessageProvIDer;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.cloud.stream.messaging.source;import org.springframework.integration.support.MessageBuilder;import org.springframework.messaging.MessageChannel;import javax.annotation.Resource;import java.util.UUID;/** * @Author EiletXIE * @Since 2020/3/14 14:13 */@EnableBinding(Source.class) //定义消息的推送管道public class MessageProvIDerImpl implements IMessageProvIDer { @Resource private MessageChannel output; // 消息发送管道 @OverrIDe public String send() { String serial = UUID.randomUUID().toString(); output.send(MessageBuilder.withPayload(serial).build()); System.out.println("*****serial: " +serial); return null; }}
Controller@RestControllerpublic class SendMessageController { @Resource private IMessageProvIDer messageProvIDer; @GetMapPing("/sendMessage") public String sendMessage() { return messageProvIDer.send(); }}
测试启动7001eureka
启动rabbitmq
rabbitmq-plugins enbale rabbitmq_managementhttp://localhost:15672/启动8801
访问:http://localhost:8801/sendMessage
消息驱动之消费者新建Modulecloud-stream-rabbitmq-consumer8802
POM<dependencIEs> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-starter-web</artifactID> </dependency> <!--监控--> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-starter-actuator</artifactID> </dependency> <!--eureka clIEnt--> <dependency> <groupID>org.springframework.cloud</groupID> <artifactID>spring-cloud-starter-netflix-eureka-clIEnt</artifactID> </dependency> <!--stream rabbit --> <dependency> <groupID>org.springframework.cloud</groupID> <artifactID>spring-cloud-starter-stream-rabbit</artifactID> </dependency> <!--热部署--> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-devtools</artifactID> <scope>runtime</scope> <optional>true</optional> </dependency> <dependency> <groupID>org.projectlombok</groupID> <artifactID>lombok</artifactID> <optional>true</optional> </dependency> <dependency> <groupID>org.springframework.boot</groupID> <artifactID>spring-boot-starter-test</artifactID> <scope>test</scope> </dependency></dependencIEs>
YMLserver: port: 8802spring: application: name: cloud-stream-consumer cloud: stream: binders: #在此处配置要绑定的rabbitmq的服务信息 defaultRabbit: #表示定义的名称,用于binding整合 type: rabbit #消息组件类型 environment: #设置rabbitmq的相关环境配置 spring: rabbitmq: host: localhost port: 5672 username: guest password: guest bindings: #服务的整合处理 input: #这个名字是一个通道的名称 destination: studyExchange #表示要使用的Exchange名称定义 content-type: application/Json #设置消息类型,本次为Json,本文要设置为“text/plain” binder: defaultRabbit #设置要绑定的消息服务的具体设置eureka: clIEnt: service-url: defaultZone: http://localhost:7001/eureka instance: lease-@R_475_5026@-interval-in-seconds: 2 #设置心跳的时间间隔(默认是30S) lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s instance-ID: receive-8802.com #在信息列表时显示主机名称 prefer-ip-address: true #访问的路径变为IP地址
主启动类@SpringBootApplicationpublic class StreamMQMain8802 { public static voID main(String[] args) { SpringApplication.run(StreamMQMain8802.class,args); }}
业务类@Component@EnableBinding(Sink.class)public class ReceiverMessageListenerController { @Value("${server.port}") private String serverPort; @StreamListener(Sink.input) public voID input(Message<String> message) { System.out.println("消费者1号, -----> 接受到的消息: " + message.getPayload() + "\t port: " + serverPort); }}
分组消费与持久化依照8802,clone出来一份8803启动RabbitMQ7001服务注册8801消息生产8802消息消费8803消息消费运行后有两个问题有重复消费问题消息持久化问题消费目前是8802/8803同时收到了,存在重复消费问题
http://localhost:8801/sendMessage
如何解决
分组和持久化属性group(重要)
比如在如下场景中,订单系统我们做集群部署,都会从RabbitMQ中获取订单信息,那如果一个订单同时被两个服务获取到,那么就会造成数据错误,我们得避免这种情况。
这时我们就可以使用Stream中的消息分组来解决
不同组是可以全面消费的(重复消费),
同一组内会发生竞争关系,只有其中一个可以消费分组原理
微服务应用放置于同一个group中,就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的,同一个组内会发生竞争关系,只有其中一个可以消费
8802/8803都变成不同组,group两个不同group:atguiguA,atguiguB8802修改YMLbindings: #服务的整合处理 input: #这个名字是一个通道的名称 destination: studyExchange #表示要使用的Exchange名称定义 content-type: application/Json #设置消息类型,本次为Json,本文要设置为“text/plain” binder: defaultRabbit #设置要绑定的消息服务的具体设置 group: atguiguA
在bindings下添加group属性即可
8803修改YMLbindings: #服务的整合处理 input: #这个名字是一个通道的名称 destination: studyExchange #表示要使用的Exchange名称定义 content-type: application/Json #设置消息类型,本次为Json,本文要设置为“text/plain” binder: defaultRabbit #设置要绑定的消息服务的具体设置 group: atguiguB
我们自己配置多数情况,生产者发送消息给某个具体微服务时只希望被消费一次,按照上面我们启动两个应用的例子,虽然它们同属一个应用,但是这个消息出现了被重复消费两次的情况。为了解决这个问题,在Spring Cloud Stream中提供了消费组的概念。
结论8802/8803实现了轮询分组,每次只有一个消费者,8801模块的发的消息只能被8802或8803其中一个接收到,这样避免了重复消费
8802/8803都变成相同组,group两个相同group:atguiguA8802修改YML8803修改YML结论同一个组的多个微服务实例,每次只会有一个拿到
持久化通过上述,解决了重复消费问题,再看看持久化停止8802/8803并去除掉8802分组group:atguiguA8803的分组group:atguiguA没有去掉8801先发送4条消息到rabbitmq,再试着启动8802、8803看他们是否能接收到发送过的消息先启动8802,无分组属性配置,后台没有打出来消息
再启动8803,有分组属性配置,后台打出来了MQ上的消息
总结我们发现有指定分组的服务8803,消息可以持久化,即使服务中途断开后重启仍然可以获得,而未指定分组的服务就会丢失断开期间发送到MQ的消息。
理解:因为消息发送时,会将消息发给各个分组,虽然8803程序已经停止了,但是它的分组B此时还是存在的,所以消息直接发给了分组B;(其实分组A也能收到消息)
而8802程序此时没有分组,也就是说,当8802启动时才给它分配默认分组,而此时新建的默认分组是在发送消息之后新建的,所以错过了此消息。
以上是内存溢出为你收集整理的【SpringCloud】SpringCloud Stream消息驱动全部内容,希望文章能够帮你解决【SpringCloud】SpringCloud Stream消息驱动所遇到的程序开发问题。
如果觉得内存溢出网站内容还不错,欢迎将内存溢出网站推荐给程序员好友。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)