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

本文主要是介绍SpringCloud 2020.0.4 系列之 Stream 消息广播与消息分组的实现，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1. 概述

2. 消息广播

3. 消息分组

4. 综述

5. 个人公众号

1. 概述

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

言归正传，今天我们来聊一下 SpringCloud 的 Stream 组件，SpringCloud Stream 是一个构建消息驱动微服务的框架，简单理解就是在 SpringCloud 中操作消息中间件的组件。

这里我们以 RabbitMQ 为例，介绍一下如何使用 Stream 实现消息广播和消息分组。

RabbitMQ 镜像模式集群的搭建，可参见我的另一篇文章《RabbitMQ 3.9.7 镜像模式集群的搭建》（RabbitMQ 3.9.7 镜像模式集群的搭建_追风人的博客-CSDN博客）

在早期的 SpringCloud 版本中常使用 @Input、@Output、@EnableBinding 和 @StreamListener 注解开发生产者与消费者。

官方原文：Deprecated as of 3.1 in favor of functional programming model。

SpringCloud 2020.0.4 版本中，已经不推荐这么开发了，因此这里我们也使用新的写法（函数式编程方式）开发。

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

2. 消息广播

2.1 主要依赖

        <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><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency>

2.2 消息实体类

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

2.3 生产者

    @Autowiredprivate 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 {// 消费广播消息@Beanpublic Consumer<MyMessage> myBroadcast() {  // 方法名必须与生产消息时自定义的绑定名称一致return message -> {log.info("接收广播消息：{}", message.getPayload());};}

2.5 application.yml 配置

spring:application:name: my-stream-newrabbitmq:   # RabbitMQ 配置addresses: 192.168.1.22:5672,192.168.1.12:5672,192.168.1.11:5672username: guestpassword: guestvirtual-host: /connection-timeout: 16000cloud:function:# 定义消费者，多个用分号分隔，当存在大于1个的消费者时，不定义不会生效definition: myBroadcaststream:bindings:# 广播消息myBroadcast-in-0:   # 消费者绑定名称，myBroadcast是自定义的绑定名称，in代表消费者，0是固定写法destination: my-broadcast-topic   # 对应的真实的 RabbitMQ ExchangemyBroadcast-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

自动生成的 Queue

两个实例的消费情况

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 消费者

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

3.3 application.yml 配置

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

3.4 验证分组消息

将 Stream 工程启动两个实例，调用发送分组消息接口，查看消息消费情况。

发送分组消息接口：

POST http://localhost:49000/stream/group?body=这是一条分组消息！

自动生成的 Exchange