本文共 13896 字,大约阅读时间需要 46 分钟。
博客参考学习视频: https://www.bilibili.com/video/BV18E411x7eT?from=search&seid=4388336378730572330
上一篇:
一句话: 屏蔽底层消息中间件的差异,降低切换版本,统一消息的编程模型
官网
https://spring.io/projects/spring-cloud-stream#overview
Spring Cloud Stream 是用于构建与共享 消息传递系统 连接的高度可伸缩的事件驱动微服务框架,该框架提供一个灵活的编程模型,它建立在spring的基础上,包括支持持久化的发布/订阅、消费以及分区 这三个核心概念
标准 MQ
为什么用 Cloud Stream
4.Binder
Stream 中的消息通信方式遵循了发布-订阅模式
Binder:很方便的连接中间件,屏蔽差异
Channel: 通道,是队列 Query 的一种抽象,在消息通讯系统中就是实现存储和转发的媒介,通过对 Channel 对队列进行配置
Source 和 Sink: 简答的可理解为参照对象是 Spring Cloud Stream 自身,从Stream 发布消息就是输出,接收消息就是输入
新建 Module: cloud-stream-rabbitmq-provider8801
POM
clould com.oy 1.0-SNAPSHOT 4.0.0 cloud-stream-rabbitmq-provider8801 org.springframework.boot spring-boot-starter-web org.springframework.cloud spring-cloud-starter-netflix-eureka-client org.springframework.cloud spring-cloud-starter-stream-rabbit org.springframework.boot spring-boot-starter-actuator org.springframework.boot spring-boot-devtools runtime true org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test
YML
server: port: 8801spring: application: name: cloud-stream-provider cloud: stream: binders: # 在此处配置要绑定的rabbitmq的服务信息; defaultRabbit: # 表示定义的名称,用于于binding整合 type: rabbit # 消息组件类型 environment: # 设置rabbitmq的相关的环境配置 spring: rabbitmq: host: 116.63.177.72 port: 5672 username: guest password: guest bindings: # 服务的整合处理 output: # 这个名字是一个通道的名称 destination: studyExchange # 表示要使用的Exchange名称定义 content-type: application/json # 设置消息类型,本次为json,文本则设置“text/plain” binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka: client: # 客户端进行Eureka注册的配置 service-url: defaultZone: http://localhost:7001/eureka instance: lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒) lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒) instance-id: send-8801.com # 在信息列表时显示主机名称 prefer-ip-address: true # 访问的路径变为IP地址
主启动类
@SpringBootApplicationpublic class StreamMQMain8801 { public static void main(String[] args) { SpringApplication.run(StreamMQMain8801.class, args); }}
业务类
发消息接口:
public interface IMessageProvider { public String send();}
发送消息接口实现类
@EnableBinding(Source.class) // 定义消息的推送管道public class MessionProvicerImpl 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(value = "/sendMessage") public String sendMessage() { return messageProvider.send(); }}
测试
注意
:有个大坑,视频里的application.yml使用了spring.cloud.stream.binders.defaultRabbit.environment.spring.rabbitmq.xx
来配置rabbitmq的环境如果你是用的其他服务器上的rabbitmq,比如我使用的我自己的华为云服务器然后创建docker容器来运行
rabbitmq。按照视频中的配置方式的话,启动时会试图连接两次rabbitmq程序第一次试图连接访问的就是application.yml中配置
的地址,此时已经订阅成功了但是程序还会在之后进行第二次连接,此时访问的地址就是localhost:5672,在我的环境中,我本地
没有rabbitmq环境,所以直接报IOException所以,如果是使用的自己的服务器来配置,则需要修改配置文件,将rabbitmq的配置
信息移动到application.yml中的spring节点下修改后的配置如下:
server: port: 8801spring: application: name: cloud-stream-provider rabbitmq: host: 华为云ip port: 5672 username: guest password: guest cloud: stream: binders: defaultRabbit: type: rabbit bindings: output: destination: studyExchange content-type: application/json binder: defaultRabbiteureka: client: # 客户端进行Eureka注册的配置 service-url: defaultZone: http://localhost:7001/eureka instance: lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒) lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒) instance-id: send-8801.com # 在信息列表时显示主机名称 prefer-ip-address: true # 访问的路径变为IP地址
新建 Module
cloud-stream-rabbitmq-consumer8802
POM
clould com.oy 1.0-SNAPSHOT 4.0.0 cloud-stream-rabbitmq-consumer8802 org.springframework.cloud spring-cloud-starter-stream-rabbit org.springframework.cloud spring-cloud-starter-netflix-eureka-client com.oy cloud-api-commons 1.0-SNAPSHOT org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-actuator org.springframework.boot spring-boot-devtools runtime true org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test
YML
server: port: 8802spring: application: name: cloud-stream-consumer rabbitmq: host: 华为云ip // localhost port: 5672 username: guest password: guest cloud: stream: binders: # 在此处配置要绑定的rabbitmq的服务信息; defaultRabbit: # 表示定义的名称,用于于binding整合 type: rabbit # 消息组件类型 bindings: # 服务的整合处理 input: # 这个名字是一个通道的名称 destination: studyExchange # 表示要使用的Exchange名称定义 content-type: application/json # 设置消息类型,本次为对象json,如果是文本则设置“text/plain” binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka: client: # 客户端进行 Eureka 注册的配置 service-url: defaultZone: http://localhost:7001/eureka instance: lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是 30 秒) lease-expiration-duration-in-seconds: 5 # 如果现在超过了 5 秒的间隔(默认是 90秒) instance-id: receive-8802.com # 在信息列表时显示主机名称 prefer-ip-address: true # 访问的路径变为 IP 地址
主启动
@SpringBootApplicationpublic class StreamMQMain8802 { public static void main(String[] args) { SpringApplication.run(StreamMQMain8802.class, args); }}
业务类
package com.oy.springcloud.controller;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.messaging.Message;import org.springframework.beans.factory.annotation.Value;import org.springframework.cloud.stream.annotation.StreamListener;import org.springframework.cloud.stream.messaging.Sink;import org.springframework.stereotype.Component;/** * @Author OY * @Date 2020/10/22 */@Component@EnableBinding(Sink.class)public class ReceiveMessageListenerController { @Value("${server.port}") private String serverPort; @StreamListener(Sink.INPUT) public void input(Messagemessage){ System.out.println(" 消 费 者 1 号 , 接 受 : "+message.getPayload()+"\t port:"+serverPort); }}
测试 8801 发送 8802 接收消息:
http://localhost:8801/sendMessage
依照 8802, clone 出来一份运行 8803
cloud-stream-rabbitmq-consumer8803
POM
clould com.oy 1.0-SNAPSHOT 4.0.0 cloud-stream-rabbitmq-consumer8803 org.springframework.boot spring-boot-starter-web org.springframework.cloud spring-cloud-starter-netflix-eureka-client org.springframework.cloud spring-cloud-starter-stream-rabbit org.springframework.boot spring-boot-starter-actuator org.springframework.boot spring-boot-devtools runtime true org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test
YML
server: port: 8803spring: application: name: cloud-stream-consumer rabbitmq: host: 116.63.177.72 port: 5672 username: guest password: guest cloud: stream: binders: # 在此处配置要绑定的rabbitmq的服务信息; defaultRabbit: # 表示定义的名称,用于于binding整合 type: rabbit # 消息组件类型 bindings: # 服务的整合处理 input: # 这个名字是一个通道的名称 destination: studyExchange # 表示要使用的Exchange名称定义 content-type: application/json # 设置消息类型,本次为对象json,如果是文本则设置“text/plain” binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka: client: # 客户端进行 Eureka 注册的配置 service-url: defaultZone: http://localhost:7001/eureka instance: lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是 30 秒) lease-expiration-duration-in-seconds: 5 # 如果现在超过了 5 秒的间隔(默认是 90秒) instance-id: receive-8803.com # 在信息列表时显示主机名称 prefer-ip-address: true # 访问的路径变为 IP 地址
主启动类
@SpringBootApplicationpublic class StreamMQMain8803 { public static void main(String[] args) { SpringApplication.run(StreamMQMain8803.class, args); }}
Controller
package com.oy.springcloud.controller;import org.springframework.beans.factory.annotation.Value;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.cloud.stream.annotation.StreamListener;import org.springframework.cloud.stream.messaging.Sink;import org.springframework.messaging.Message;import org.springframework.stereotype.Component;/** * @Author OY * @Date 2020/10/22 */@Component@EnableBinding(Sink.class)public class ReceiveMessageListenerController { @Value("${server.port}") private String serverPort; @StreamListener(Sink.INPUT) public void input(Messagemessage){ System.out.println(" 消 费 者 2 号 , 接 受 : "+message.getPayload()+"\t port:"+serverPort); }}
分组和持久化属性 group
原理
微服务应用放置于同一个 group 中,就能保证消息只会被其中一个应用消费一次。不同的组是可以消费的。同一个组内发生的竞争的关系,只有一个可以消费。
8802/8803 都变成不同组, group 两个不同
group: atguiguA
group: atguiguB
分布式微服务应用为了实现高可用和负载均衡,实际上都会户数多个实例,本例启动了两个消费微服务(8802/8803)多数情况,生产者发送消息给某个具体微服务只希望被消费一次,按照上面我们启动两个应用的例子,虽然它们同属一个应用,但是这个消息出现了被重复消费两次的情况。为了解决这个问题,在Spring Cloud Stream 中提供了消费组
的概念。
8802/8803 实现了轮询分组, 每次只有一个消费者 8801 模块的发的消息只能被 8802 或 8803 其中一个接收到, 这样避免了重复
消费 。
8802/8803 都变成相同组, group 两个相同 。
group: atguiguA
group: atguiguA
结论: 同一个组的多个微服务实例, 每次只会有一个拿到
通过上述, 解决了重复消费问题, 再看看持久化,停止 8802/8803 并去除掉 8802 的分组 group:atguiguA,8803 的分组
group:atguiguA 没有去掉。8801 先发送 4 条信息到 rabbitmq。
guA
2. 8802 修改 YMLgroup: atguiguA
group: atguiguA
结论: 同一个组的多个微服务实例, 每次只会有一个拿到
[外链图片转存中…(img-4X73GBwt-1603423170797)]
通过上述, 解决了重复消费问题, 再看看持久化,停止 8802/8803 并去除掉 8802 的分组 group:atguiguA,8803 的分组
group:atguiguA 没有去掉。8801 先发送 4 条信息到 rabbitmq。
转载地址:http://uabki.baihongyu.com/