SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现

1. 概述
老话说的好:事情太多,做不过来,就先把事情记在本子上,然后理清思路、排好优先级,一件一件的去完成。

言归正传,今天我们来聊一下 SpringCloud 的 Stream 组件,SpringCloud Stream 是一个构建消息驱动微服务的框架,简单理解就是在 SpringCloud 中操作消息中间件的组件。
这里我们以 RabbitMQ 为例,介绍一下如何使用 Stream 实现 消息广播 和 消息分组。
RabbitMQ 镜像模式集群的搭建,可参见我的另一篇文章《RabbitMQ 3.9.7 镜像模式集群的搭建》(https://www.cnblogs.com/w84422/p/15356202.html)

在早期的 SpringCloud 版本中常使用 @Input、@Output、@EnableBinding 和 @StreamListener 注解开发生产者与消费者。
官方原文:Deprecated as of 3.1 in favor of functional programming model。
SpringCloud 2020.0.4 版本中,已经不推荐这么开发了,因此这里我们也使用新的写法(函数式编程方式) 开发。

闲话不多说,直接上代码。

2. 消息广播
2.1 主要依赖

org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-actuator org.springframework.cloud spring-cloud-starter-stream-rabbit


2.2 消息实体类
@Setter @Getter public class MyMessage implements java.io.Serializable {// 消息体 private String payload; }


2.3 生产者
@Autowired private StreamBridge streamBridge; // 发送广播消息 @PostMapping("/broadcast") public String sendBroadcastMessage(@RequestParam("body") String body) {MyMessage myMessage = new MyMessage(); myMessage.setPayload(body); // 生产消息 // 第一个参数是绑定名称,格式为:自定义的绑定名称-out-0,myBroadcast是自定义的绑定名称,out代表生产者,0是固定写法 // 自定义的绑定名称必须与消费方法的方法名保持一致 // 第二个参数是发送的消息实体 streamBridge.send("myBroadcast-out-0", myMessage); return "SUCCESS"; }


2.4 消费者
@Slf4j @Component // 消费者 public class Concumer {// 消费广播消息 @Bean public Consumer myBroadcast() {// 方法名必须与生产消息时自定义的绑定名称一致return message -> { log.info("接收广播消息:{}", message.getPayload()); }; }


2.5 application.yml 配置
spring: application: name: my-stream-new rabbitmq:# RabbitMQ 配置 addresses: 192.168.1.22:5672,192.168.1.12:5672,192.168.1.11:5672 username: guest password: guest virtual-host: / connection-timeout: 16000 cloud: function: # 定义消费者,多个用分号分隔,当存在大于1个的消费者时,不定义不会生效 definition: myBroadcast stream: bindings:# 广播消息 myBroadcast-in-0:# 消费者绑定名称,myBroadcast是自定义的绑定名称,in代表消费者,0是固定写法 destination: my-broadcast-topic# 对应的真实的 RabbitMQ Exchange myBroadcast-out-0:# 生产者绑定名称,myBroadcast是自定义的绑定名称,out代表生产者,0是固定写法 destination: my-broadcast-topic# 对应的真实的 RabbitMQ Exchange


注意:spring.cloud.function.definition 属性非常重要,务必配置

2.6 验证广播消息
将 Stream 工程 启动两个实例,调用发送广播消息接口,查看消息消费情况。
发送广播消息接口:
POST http://localhost:49000/stream/broadcast?body=这是一条广播消息!

自动生成的 Exchange
SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片


自动生成的 Queue
SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片


两个实例的消费情况
SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片


SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片


3. 消息分组
消息分组后,同一分组的消费服务,只能有一个消费者消费到消息。
3.1 生产者
// 发送分组消息 @PostMapping("/group") public String sendGroupMessage(@RequestParam("body") String body) {MyMessage myMessage = new MyMessage(); myMessage.setPayload(body); // 生产消息 streamBridge.send("myGroup-out-0", myMessage); return "SUCCESS"; }


3.2 消费者
// 消费分组消息 @Bean public Consumer myGroup() {// 方法名必须与生产消息时自定义的绑定名称一致return message -> { log.info("接收分组消息:{}", message.getPayload()); }; }


3.3 application.yml 配置
spring: application: name: my-stream-new rabbitmq:# RabbitMQ 配置 addresses: 192.168.1.22:5672,192.168.1.12:5672,192.168.1.11:5672 username: guest password: guest virtual-host: / connection-timeout: 16000 cloud: function: # 定义消费者,多个用分号分隔,当存在大于1个的消费者时,不定义不会生效 definition: myBroadcast; myGroup stream: bindings: # 广播消息 myBroadcast-in-0:# 消费者绑定名称,myBroadcast是自定义的绑定名称,in代表消费者,0是固定写法 destination: my-broadcast-topic# 对应的真实的 RabbitMQ Exchange myBroadcast-out-0:# 生产者绑定名称,myBroadcast是自定义的绑定名称,out代表生产者,0是固定写法 destination: my-broadcast-topic# 对应的真实的 RabbitMQ Exchange# 分组消息 myGroup-in-0: # 消费者绑定名称 destination: my-group-topic# 对应的真实的 RabbitMQ Exchange group: my-group-1# 同一分组的消费服务,只能有一个消费者消费到消息 myGroup-out-0: # 生产者绑定名称 destination: my-group-topic# 对应的真实的 RabbitMQ Exchange


3.4 验证分组消息
将 Stream 工程 启动两个实例,调用发送分组消息接口,查看消息消费情况。
发送分组消息接口:
POST http://localhost:49000/stream/group?body=这是一条分组消息!

自动生成的 Exchange
SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片



自动生成的 Queue
SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片



两个实例的消费情况
SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片



SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片



4. 综述
今天聊了一下 SpringCloud Stream 组件中 消息广播 与 消息分组 的实现 ,希望可以对大家的工作有所帮助。
欢迎帮忙点赞、评论、转发、加关注 :)
关注追风人聊Java,每天更新Java干货。

5. 个人公众号
追风人聊Java,欢迎大家关注
SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现
文章图片

【SpringCloud|SpringCloud 2020.0.4 系列之 Stream 消息广播 与 消息分组 的实现】

    推荐阅读