SpringCloud系列十一：SpringCloudStream（SpringCloudStream 简介、创建消息生产者、创建消息消费者、自定义消息通道、分组与持久化、设置 RoutingKey）

淩亂°似流年 2022-05-22 10:29 290阅读 0赞

非常感谢[https://www.cnblogs.com/leeSmall/p/8900518.html][https_www.cnblogs.com_leeSmall_p_8900518.html]

Springboot 技术学习  [https://www.itkc8.com][https_www.itkc8.com]

1、概念：SpringCloudStream

## 2、具体内容 ##

### 2.1、SpringCloudStream 简介 ###

SpringCloudStream 就是使用了基于消息系统的微服务处理架构。对于消息系统而言一共分为两类：基于应用标准的 JMS、基于协议标准的 AMQP，在整个 SpringCloud 之中支持有 RabbitMQ、Kafka 组件的消息系统。利用 SpringCloudStream 可以实现更加方便的消息系统的整合处理，但是推荐还是基于 RabbitMQ 实现会更好一些。

为什么 SpringCloud 中要提供有一个类似于消息驱动的 SpringCloudStream 呢？

如果通过 Java 历史上的分布式的开发架构大家不难发现，对于消息系统，实际上最初的 SUN 公司是非常看中的，所以在 EJB 的时代里面专门提供有消息驱动 Bean（Message Driven Bean、MDB）利用消息驱动 Bean 可以进行消息的处理操作。利用消息驱动 bean 的模式可以简化用户的操作复杂度，直接传递一些各类的数据即可实现业务的处理操作。

于是在 SpringBoot 的之中为了方便开发者去整合消息组件，也提供有一系列的处理支持，但是如果按照这些方式来在 SpringCloud 之中进行消息处理，有些人会认为比较麻烦，所以在 SpringCloud 里面将消息整合的处理操作进行了进一步的抽象操作， 实现了更加简化的消息处理。

总结：SpringCloudStream 就是实现了 MDB 功能，同时可以更加简化方便的整合消息组件。

SpringCloudStream的工作原理：

![1227483-20180421101011343-410776027.png][]

说明：最底层是消息服务，中间层是绑定层，绑定层和底层的消息服务进行绑定，顶层是消息生产者和消息消费者，顶层可以向绑定层生产消息和和获取消息消费

###  2.2、创建消息生产者 ###

本次基于 RabbitMQ 实现消息的生产者的微服务操作，在整个的生产者项目之中，首先创建了一个新的 Maven 模块： microcloud-stream-provider-8401。

1、 【microcloud-stream-provider-8401】修改 pom.xml 配置文件，在这个配置文件之中要追加如下的依赖程序包：

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

2、 【microcloud-stream-provider-8401】修改 application.yml 配置文件，追加如下的绑定处理配置：

![复制代码][copycode.gif]

server:
      port: 8401
    eureka: 
      client: # 客户端进行Eureka注册的配置
        service-url:
          defaultZone: http://edmin:studyjava@eureka-7001.com:7001/eureka,http://edmin:studyjava@eureka-7002.com:7002/eureka,http://edmin:studyjava@eureka-7003.com:7003/eureka
      instance:
        lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
        lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
        instance-id: send-8401.com  # 在信息列表时显示主机名称
        prefer-ip-address: true     # 访问的路径变为IP地址
    spring:
      cloud:
        stream:
          binders: # 在此处配置要绑定的rabbitmq的服务信息；
            defaultRabbit: # 表示定义的名称，用于于binding整合
              type: rabbit # 消息组件类型
              environment: # 设置rabbitmq的相关的环境配置
                spring:
                  rabbitmq:
                  addresses: rabbitmq-server
                  username: studyjava
                  password: hello
                  virtual-host: /
          bindings: # 服务的整合处理
            output: # 这个名字是一个通道的名称，在分析具体源代码的时候会进行说明
              destination: studyExchange # 表示要使用的Exchange名称定义
              content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
              binder: defaultRabbit # 设置要绑定的消息服务的具体设置
      application:
        name: microcloud-stream-provider

![复制代码][copycode.gif]

3、 修改 hosts 配置文件，追加主机映射：

127.0.0.1 send-8401.com

4、 【microcloud-stream-provider-8401】定义一个消息的发送接口：

![复制代码][copycode.gif]

package cn.study.microcloud.service;
    
    import cn.study.vo.Company;
    
    public interface IMessageProvider {
    
        /**
        * 实现消息的发送，本次发送的消息是一个对象（自动变为json）
        * @param company VO对象，该对象不为null*/
          public void send(Company company) ;
    }

![复制代码][copycode.gif]

5、 【microcloud-stream-provider-8401】定义消息发送的实现子类：

![复制代码][copycode.gif]

package cn.study.microcloud.service.impl;
    
    import javax.annotation.Resource;
    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 cn.study.microcloud.service.IMessageProvider;
    import cn.study.vo.Company;
    
    @EnableBinding(Source.class) // 可以理解为是一个消息的发送管道的定义
    public class MessageProviderImpl implements IMessageProvider {
        @Resource
        private MessageChannel output; // 消息的发送管道
    
        @Override
        public void send(Company company) {
            this.output.send(MessageBuilder.withPayload(company).build()); // 创建并发送消息
        }
    }

![复制代码][copycode.gif]

6、 【microcloud-stream-provider-8401】分析一下 Source 类的源代码：

![复制代码][copycode.gif]

public interface Source {
    
        String OUTPUT = "output"; // 之前所设置的消息发送的管道
    
        @Output(Source.OUTPUT)
        MessageChannel output();
    
    }

![复制代码][copycode.gif]

7、 【microcloud-stream-provider-8401】定义程序主类：

![复制代码][copycode.gif]

package cn.study.microcloud;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
    import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
    import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
    @SpringBootApplication
    @EnableEurekaClient
    @EnableCircuitBreaker
    @EnableDiscoveryClient
    public class StreamProvider_8401_StartSpringCloudApplication {
        public static void main(String[] args) {
            SpringApplication.run(StreamProvider_8401_StartSpringCloudApplication.class, args);
        }
    }

![复制代码][copycode.gif]

8、 【microcloud-stream-provider-8401】编写测试类：

· 保证你的 pom.xml 文件之中存在有测试的依赖程序包：

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>

· 编写具体的测试程序类：

![复制代码][copycode.gif]

package cn.study.microcloud.test;
    
    import javax.annotation.Resource;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    import org.springframework.test.context.web.WebAppConfiguration;
    
    import cn.study.microcloud.StreamProvider_8401_StartSpringCloudApplication;
    import cn.study.microcloud.service.IMessageProvider;
    import cn.study.vo.Company;
    
    @RunWith(SpringJUnit4ClassRunner.class)
    @SpringBootTest(classes = StreamProvider_8401_StartSpringCloudApplication.class)
    @WebAppConfiguration
    public class TestMessageProvider {
        @Resource
        private IMessageProvider messageProvider;
    
        @Test
        public void testSend() {
            Company company = new Company();
            company.setTitle("studyjava");
            company.setNote("更多资源请登录：www.study.cn");
            this.messageProvider.send(company); // 消息发送
        }
    }

![复制代码][copycode.gif]

9、 启动 RabbitMQ 以及相应的微服务进行消息的发送处理，如果可以检测到 RabbitMQ 上的活动信息就表示该微服务创建成功。

![1227483-20180421110543861-8676497.png][]

###  2.3、创建消息消费者 ###

在之前已经成功的实现了消息的发送处理，但是这个消息由于只是一个临时消息并且只是发送到了 RabbitMQ 之中，那么现在 如果要想进行该消息的接收就必须通过 RabbitMQ 获取消息内容。

1、 【microcloud-stream-consumer-8402】通过“microcloud-stream-provider-8401”模块复制本模块；

2、 【microcloud-stream-consumer-8402】一定要保证 pom.xml 文件之中拥有如下的依赖包：

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

3、 修改 hosts 主机映射，追加新的主机名称：

127.0.0.1 receive-8402.com

4、 【microcloud-stream-consumer-8402】修改 application.yml 配置文件：

![复制代码][copycode.gif]

server:
      port: 8402
    eureka: 
      client: # 客户端进行Eureka注册的配置
        service-url:
          defaultZone: http://edmin:studyjava@eureka-7001.com:7001/eureka,http://edmin:studyjava@eureka-7002.com:7002/eureka,http://edmin:studyjava@eureka-7003.com:7003/eureka
      instance:
        lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
        lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
        instance-id: receive-8402.com  # 在信息列表时显示主机名称
        prefer-ip-address: true     # 访问的路径变为IP地址
    spring:
      cloud:
        stream:
          binders: # 在此处配置要绑定的rabbitmq的服务信息；
            defaultRabbit: # 表示定义的名称，用于于binding整合
              type: rabbit # 消息组件类型
              environment: # 设置rabbitmq的相关的环境配置
                spring:
                  rabbitmq:
                  addresses: rabbitmq-server
                  username: studyjava
                  password: hello
                  virtual-host: /
          bindings: # 服务的整合处理
            input: # 这个名字是一个通道的名称，在分析具体源代码的时候会进行说明
              destination: studyExchange # 表示要使用的Exchange名称定义
              content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
              binder: defaultRabbit # 设置要绑定的消息服务的具体设置
      application:
        name: microcloud-stream-consumer

![复制代码][copycode.gif]

5、 【microcloud-stream-consumer-8402】定义一个消息的监听程序类：

![复制代码][copycode.gif]

package cn.study.microcloud.service;
    
    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;
    import cn.study.vo.Company;
    
    @Component
    @EnableBinding(Sink.class)
    public class MessageListener {
        @StreamListener(Sink.INPUT)
        public void input(Message<Company> message) {
            System.err.println("【*** 消息接收 ***】" + message.getPayload());
        }
    }

![复制代码][copycode.gif]

6、 【microcloud-stream-consumer-8402】观察 Sink 源代码：

![复制代码][copycode.gif]

public interface Sink {
    
        String INPUT = "input";
    
        @Input(Sink.INPUT)
        SubscribableChannel input();
    
    }

![复制代码][copycode.gif]

7、 首先启动消息的消费端，而后再启动消息的生产者发送消息。

![1227483-20180421112645465-719957084.png][]

那么此时实现了了一个基于 RabbitMQ 定义的 SpringCloudStream 基本操作功能。

###  2.4、自定义消息通道 ###

现在已经实现了一个基础的 SpringCloudStream 处理操作，但是在本次操作之中一直使用的都是系统中提供好的 Source （output）、Sink（input），如果说现在用户有需要也可以定义自己的通道名称。

1、 【micocloud-api】由于现在有两个模块都需要使用到自定义消息通道的配置，所以应该将这个配置定义为一个公共的程序处理 类，修改 pom.xml 配置文件，引入相应的开发包：

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

2、 【micocloud-api】使用一个自定义的通道：

![复制代码][copycode.gif]

package cn.study.channel;
    
    import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.SubscribableChannel;
    
    public interface DefaultProcess {
        public static final String OUTPUT = "study_output"; // 输出通道名称
        public static final String INPUT = "study_input"; // 输入通道名称
    
        @Input(DefaultProcess.INPUT)
        public SubscribableChannel input();
    
        @Output(DefaultProcess.OUTPUT)
        public MessageChannel output();
    }

![复制代码][copycode.gif]

3、 【microcloud-stream-provider-8401】修改 application.yml 配置文件：

![复制代码][copycode.gif]

spring:
      cloud:
        stream:
          binders: # 在此处配置要绑定的rabbitmq的服务信息；
            defaultRabbit: # 表示定义的名称，用于于binding整合
              type: rabbit # 消息组件类型
              environment: # 设置rabbitmq的相关的环境配置
                spring:
                  rabbitmq:
                  addresses: rabbitmq-server
                  username: studyjava
                  password: hello
                  virtual-host: /
          bindings: # 服务的整合处理
            study_output: # 这个名字是一个通道的名称，在分析具体源代码的时候会进行说明
              destination: studyExchange # 表示要使用的Exchange名称定义
              content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
              binder: defaultRabbit # 设置要绑定的消息服务的具体设置

![复制代码][copycode.gif]

4、 【microcloud-stream-consumer-8402】修改 application.yml 配置文件：

![复制代码][copycode.gif]

spring:
      cloud:
        stream:
          binders: # 在此处配置要绑定的rabbitmq的服务信息；
            defaultRabbit: # 表示定义的名称，用于于binding整合
              type: rabbit # 消息组件类型
              environment: # 设置rabbitmq的相关的环境配置
                spring:
                  rabbitmq:
                  addresses: rabbitmq-server
                  username: studyjava
                  password: hello
                  virtual-host: /
          bindings: # 服务的整合处理
            study_input: # 这个名字是一个通道的名称，在分析具体源代码的时候会进行说明
              destination: studyExchange # 表示要使用的Exchange名称定义
              content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
              binder: defaultRabbit # 设置要绑定的消息服务的具体设置

![复制代码][copycode.gif]

5、 【microcloud-stream-provider-8401】修改消息的发送子类：

![复制代码][copycode.gif]

package cn.study.microcloud.service.impl;
    
    import javax.annotation.Resource;
    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 cn.study.channel.DefaultProcess;
    import cn.study.microcloud.service.IMessageProvider;
    import cn.study.vo.Company;
    
    @EnableBinding(DefaultProcess.class) // 可以理解为是一个消息的发送管道的定义
    public class MessageProviderImpl implements IMessageProvider {
        @Resource
        private MessageChannel output; // 消息的发送管道
    
        @Override
        public void send(Company company) {
            this.output.send(MessageBuilder.withPayload(company).build()); // 创建并发送消息
        }
    }

![复制代码][copycode.gif]

6、 【microcloud-stream-consumer-8402】修改 MessageListener 程序类：

![复制代码][copycode.gif]

package cn.study.microcloud.service;
    
    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;
    
    import cn.study.channel.DefaultProcess;
    import cn.study.vo.Company;
    
    @Component
    @EnableBinding(DefaultProcess.class)
    public class MessageListener {
        @StreamListener(Sink.INPUT)
        public void input(Message<Company> message) {
            System.err.println("【*** 消息接收 ***】" + message.getPayload());
        }
    }

![复制代码][copycode.gif]

7、 随后就可以使用自定义的新的通道名称进行 Stream 处理操作了。

###  2.5、分组与持久化 ###

在上面的程序里面成功的实现了消息的发送以及接收，但是需要注意一个问题，所发送的消息在默认情况下它都属于一种临时消息，也就是说如果现在没有消费者进行消费处理，那么该消息是不会被保留的。

如果要想实现持久化的消息处理，重点在于消息的消费端配置，同时也需要考虑到一个分组的情况（有分组就表示该消息可以进行持久化）。

1、 【microcloud-stream-consumer-8402】修改 application.yml 配置文件，追加分组配置：

![复制代码][copycode.gif]

spring:
      cloud:
        stream:
          binders: # 在此处配置要绑定的rabbitmq的服务信息；
            defaultRabbit: # 表示定义的名称，用于于binding整合
              type: rabbit # 消息组件类型
              environment: # 设置rabbitmq的相关的环境配置
                spring:
                  rabbitmq:
                  addresses: rabbitmq-server
                  username: studyjava
                  password: hello
                  virtual-host: /
          bindings: # 服务的整合处理
            study_input: # 这个名字是一个通道的名称，在分析具体源代码的时候会进行说明
              destination: studyExchange # 表示要使用的Exchange名称定义
              content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
              binder: defaultRabbit # 设置要绑定的消息服务的具体设置
              group: study-group # 进行操作的分组，实际上就表示持久化

![复制代码][copycode.gif]

在 SpringCloudStream 之中如果要设置持久化队列，则名称为“destination.group”。此时关闭掉消费端的微服务之后该队列信息依然会被保留在 RabbitMQ 之中。而后在关闭消费端的情况下去运行消息生产者，发送完消息后再运行消息的消费端仍然可以接收到之前的消息。

###  2.6、设置 RoutingKey ###

默认情况下之前的程序都是属于广播消息，也就是说所有的消费者都可以接收发送消息内容，在 RabbitMQ 里面支持有直连消息，而直连消息主要是通过 RoutingKey 来实现，利用直连消息可以实现准确的消息消费端的接收处理。

1、 【microcloud-stream-consumer-8402】修改 application.yml 配置文件：

![复制代码][copycode.gif]

server:
      port: 8402
    eureka: 
      client: # 客户端进行Eureka注册的配置
        service-url:
          defaultZone: http://edmin:studyjava@eureka-7001.com:7001/eureka,http://edmin:studyjava@eureka-7002.com:7002/eureka,http://edmin:studyjava@eureka-7003.com:7003/eureka
      instance:
        lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
        lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
        instance-id: receive-8402.com  # 在信息列表时显示主机名称
        prefer-ip-address: true     # 访问的路径变为IP地址
    spring:
      cloud:
        stream:
          rabbit: # 进行rabbit的相关绑定配置
            bindings:
              study_input:
                consumer: # 进行消费端配置
                  bindingRoutingKey: study-key # 设置一个RoutingKey信息
          binders: # 在此处配置要绑定的rabbitmq的服务信息；
            defaultRabbit: # 表示定义的名称，用于于binding整合
              type: rabbit # 消息组件类型
              environment: # 设置rabbitmq的相关的环境配置
                spring:
                  rabbitmq:
                  addresses: rabbitmq-server
                  username: studyjava
                  password: hello
                  virtual-host: /
          bindings: # 服务的整合处理
            study_input: # 这个名字是一个通道的名称，在分析具体源代码的时候会进行说明
              destination: RKExchange # 表示要使用的Exchange名称定义
              content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
              binder: defaultRabbit # 设置要绑定的消息服务的具体设置
              group: study-group # 进行操作的分组，实际上就表示持久化
              
      application:
        name: microcloud-stream-consumer

![复制代码][copycode.gif]

2、 【microcloud-stream-provider-8401】定义 RoutingKey 的表达式配置：

![复制代码][copycode.gif]

server:
      port: 8401
    eureka: 
      client: # 客户端进行Eureka注册的配置
        service-url:
          defaultZone: http://edmin:studyjava@eureka-7001.com:7001/eureka,http://edmin:studyjava@eureka-7002.com:7002/eureka,http://edmin:studyjava@eureka-7003.com:7003/eureka
      instance:
        lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
        lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
        instance-id: send-8401.com  # 在信息列表时显示主机名称
        prefer-ip-address: true     # 访问的路径变为IP地址
    spring:
      cloud:
        stream:
          rabbit: # 进行rabbit的相关绑定配置
            bindings:
              study_output:
                producer: # 进行消费端配置
                  routing-key-expression: '''study-key''' #定义 RoutingKey 的表达式配置
          binders: # 在此处配置要绑定的rabbitmq的服务信息；
            defaultRabbit: # 表示定义的名称，用于于binding整合
              type: rabbit # 消息组件类型
              environment: # 设置rabbitmq的相关的环境配置
                spring:
                  rabbitmq:
                  addresses: rabbitmq-server
                  username: studyjava
                  password: hello
                  virtual-host: /
          bindings: # 服务的整合处理
            study_output: # 这个名字是一个通道的名称，在分析具体源代码的时候会进行说明
              destination: RKExchange # 表示要使用的Exchange名称定义
              content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
              binder: defaultRabbit # 设置要绑定的消息服务的具体设置
      application:
        name: microcloud-stream-provider

![复制代码][copycode.gif]

3、 首先运行消费端程序，随后在运行生产端，只有 RoutingKey 匹配了之后才可以正常进行消息的接收处理。

![1227483-20180421120303433-197582423.png][]

Springboot 技术学习  [https://www.itkc8.com][https_www.itkc8.com]

[https_www.cnblogs.com_leeSmall_p_8900518.html]: https://www.cnblogs.com/leeSmall/p/8900518.html
[https_www.itkc8.com]: https://www.itkc8.com/
[1227483-20180421101011343-410776027.png]: /images/20220522/a6b1267e889f41e6874e954c3737e123.png
[copycode.gif]: /images/20220522/dedff9dfb2b6438ab85a39353401196f.png
[1227483-20180421110543861-8676497.png]: /images/20220522/4b7953fae2944b19954bee75556965e5.png
[1227483-20180421112645465-719957084.png]: /images/20220522/ddbc786a74ff45928ecfd70667e80d40.png
[1227483-20180421120303433-197582423.png]: /images/20220522/3bc9850a19d248fcb034d23dc44dd750.png