Spring Cloud（十八）：Stream消息驱动

怼烎@ 2022-11-15 11:53 219阅读 0赞

## 1. 概述 ##

*  官方定义Spring Cloud Stream 是一个构件消息驱动微服务的框架。
 *  应用程序通过inputs或者outputs来与Spring Cloud Stream中的binder对象交互，通过我们配置来binding（绑定），二Spring Cloud Stream的binder对象负责与消息中间件交互，所以，我们只需要搞清楚如何与Spring Cloud Stream 交互就可以方便使用消息驱动的方式。
 *  通过使用Spring integration来连接消息代理中间件以实现消息时间驱动.
 *  Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。
 *  目前仅支持RabbitMq、Kafka。
 *  简单来说就是，可以屏蔽底层消息中间件的差异，降低切换版本，统一消息的编程模型。

## 2. 官网 ##

https://spring.io/projects/spring-cloud-stream\#overview  
Spring Cloud Stream 是用于构建与共享消息传递系统链接的高度可伸缩的事件驱动微服务框架，该框架提供了一个灵活的编程模型，它建立在已经建立和熟悉的Spring和最佳实践上，包括支持持久化的发布/订阅、消费组以及消费分区这三个核心概念。  
![在这里插入图片描述][20210401070131653.png]  
https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/

Spring Cloud Stream中文指导手册  
https://m.wang1314.com/doc/webapp/topic/20971999.html

## 3. 设计思想 ##

**标准MQ**  
![在这里插入图片描述][20210401070421954.png]

*  生产者/消费者之间靠消息媒介传递信息内容（Message）
 *  消息必须走特定的通道（消息通道MessageChannel）
 *  消息通道里的消息如何被消费呢，谁负责收发处理  
    消息通道MessageChannel的子接口SubscribableChannel,由MessageHandler消息处理器订阅

**为什么用Cloud Stream**  
比方说我们用到了RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，像RabbitMQ有exchange，kafka有Topic和Partitions分区。  
![在这里插入图片描述][20210401070738845.png]

这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了两个消息队列的其中一种，后面的业务需求，我们想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都要重新推到重新做，因为他跟我们的系统耦合了，这时候Spring Cloud Stream 给我们提供了一种解耦合的方式。

stream凭什么可以统一底层差异  
在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，他们的实现细节上会有较大的差异性，通过定义绑定器作为中间层，完美的实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel通道，使得应用程序不需要在考虑各种不同的消息中间件实现。  
通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离。

*  Binder（INPUT对应于消费者），（OUTPUT对应于生产者）

Stream对消息中间件的进一步封装，可以做到代码层面对中间件的无感知，甚至于动态的切换中间件（rabbitMq和kafka），使得微服务开发的高度解耦，服务可以关注更多自己的业务流程。

![在这里插入图片描述][20210401070938251.png]  
通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离。

**Stream中的消息通信方式遵循了发布-订阅模式**

*  topic主题进行广播  
    在RabbitMQ就是Exchange  
    在kafka中就是Topic

## 4. Spring Cloud Stream标准流程套路 ##

![在这里插入图片描述][20210401071105973.png]  
![在这里插入图片描述][20210401071125318.png]  
**Binder**  
很方便的连接中间件，屏蔽差异

**Channel**  
通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过对Channel对队列进行配置。

**Source和Sink**  
简单的可理解为参照对象是Spring Cloud Stream自身，从Stream发布消息就是输出，接受消息就是输入。

**编码API和常用注解**  
![在这里插入图片描述][20210401071307919.png]

## 5. 案例说明 ##

**工程中新建三个子模块**

*  cloud-stream-rabbitmq-provider8801,作为生产者进行发消息模块
 *  cloud-stream-rabbitmq-consumer8802,作为消息接收模块
 *  cloud-stream-rabbitmq-consumer8803,作为消息接收模块

### 5.1 消息驱动之生产者 ###

**新建Module**  
cloud-stream-rabbitmq-provider8801

**pom**

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

**yml**

server:
      port: 8801
    
    spring:
      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: #客户端进行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地址

ps：binder: defaultRabbit 报红无所谓 把  
![在这里插入图片描述][2021040409305610.png]  
这个调一下就行了

**主启动类**

package com.lele.springcloud;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /**
     * @author: lele
     * @date: 2021/4/4 9:43
     * @description:
     */
    @SpringBootApplication
    public class StreamMQMain8801 {
        
    
        public static void main(String[] args) {
        
            SpringApplication.run(StreamMQMain8801.class, args);
        }
    }

**controller**

package com.lele.springcloud.controller;
    
    import com.lele.springcloud.service.IMessageProvider;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    
    /**
     * @author: lele
     * @date: 2021/4/4 10:14
     * @description:
     */
    @RestController
    public class SendMessageController {
        
    
        @Resource
        private IMessageProvider messageProvider;
    
        @GetMapping(value = "/sendMessage")
        public String sendMessage() {
        
            return messageProvider.send();
        }
    }

**service**

package com.lele.springcloud.service;
    
    /**
     * @author: lele
     * @date: 2021/4/4 9:49
     * @description:
     */
    public interface IMessageProvider {
        
    
        public String send();
    }

**impl**

package com.lele.springcloud.service.impl;
    
    import com.lele.springcloud.service.IMessageProvider;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.messaging.Source;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.support.MessageBuilder;
    
    import javax.annotation.Resource;
    import java.util.UUID;
    
    /**
     * @author: lele
     * @date: 2021/4/4 9:53
     * @description:
     */
    @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;
        }
    }

**测试**

*  启动7001eureka
 *  启动rabbitmq
 *  启动8801

访问：http://localhost:8801/sendMessage

### 5.2 消费驱动之消费者 ###

**新建Module**  
cloud-stream-rabbitmq-consumer8802

**pom**

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <parent>
            <artifactId>spring-cloud-demo</artifactId>
            <groupId>com.lele.springcloud</groupId>
            <version>1.0-SNAPSHOT</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>cloud-stream-rabbitmq-consumer8802</artifactId>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
            </dependency>
            <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.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>
    </project>

**yml**

server:
      port: 8802
    
    spring:
      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-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30秒）
        lease-expiration-duration-in-seconds: 5 #如果现在超过了5秒的间隔（默认是90秒）
        instance-id: receive-8802.com #在信息列表时显示主机名称
        prefer-ip-address: true  #访问的路径变为ip地址

**主启动类StreamMQMain8802**

package com.lele.springcloud;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /**
     * @author: lele
     * @date: 2021/4/4 12:31
     * @description:
     */
    @SpringBootApplication
    public class StreamMain8802 {
        
        public static void main(String[] args) {
        
            SpringApplication.run(StreamMain8802.class, args);
        }
    }

**业务类**

package com.lele.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;
    
    import javax.annotation.Resource;
    
    /**
     * @author: lele
     * @date: 2021/4/4 12:35
     * @description:
     */
    @Component
    @EnableBinding(Sink.class)
    public class ReceiveMessageListenerController {
        
    
        @Value("${server.port}")
        private String serverPort;
    
        @StreamListener(Sink.INPUT)
        public void input(Message<String> message) {
        
            System.out.println("消费者1号，------>接收到的消息："+message.getPayload()+"\t port:"+serverPort);
        }
    }

访问：http://localhost:8801/sendMessage

### 5.3 分组消费与持久化 ###

**依照8802，clone出来一份运行8803**  
cloud-stream-rabbitmq-consumer8803

**启动**  
RabbitMQ  
7001 服务注册  
8801 消息生产  
8802 消息消费  
8803 消息消费

运行后两个问题：  
有重复消费问题

**消费**  
**目前是8802/8803同时都收到了，存在重复消费问题**  
比如在如下场景中，订单系统我们做集群部署，都会从RebbitMQ中获取订单信息，那如果一个订单同时被两个服务获取到，那么就会造成数据错误，我们得避免这种情况，这是我们就可以使用Stream中的消息分组来解决。

![在这里插入图片描述][20210404132026860.png]  
注意在Stream中处于同一个group中的多个消费者都是竞争关系，就能够保证消息只会被其中一个应用消费一次，不同组是可以全面消费的（重复消费）

**分组**  
微服务应用放置于同一个group中，就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的，同一个组内会发生竞争关系，只有其中一个可以消费。

*  8802/8803都变成不同组，group两个不同

8802修改YML  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70]

8803修改YML  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70 1]

分布式微服务应用为了实现高可用和负载均衡，实际上都会部署多个实例，本次启动了两个消费微服务（8802/8803）

多数情况，生产者发送消息给某个具体微服务时只希望被消费一次，按照上面我们启动两个应用的例子，虽然它们同属一个应用，但是这个消息出现了被重复消费两次的情况，为了解决这个问题，在Spring Cloud Stream 中提供了消费组的观念。

*  8802/8803都变成相同组，group两个相同  
    8802修改YML  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70 2]  
    8803修改YML  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70 3]

同一个组的多个微服务实例，每次只会有一个拿到8802/8803实现了轮询分组，每次只有一个消费者 8801模块的发的消息只能被8802或8803其中一个接收到，这样避免了重复消费。

**持久化**  
停止8802/8803并去除掉8802的分组group: leleA

8801先发送4条信息到rabbitmq

启动8802，无分组属性配置，后台没有打出来消息

启动8803，有分组属性配置，后台打出来了MQ上的消息

[20210401070131653.png]: /images/20221022/10437291e7c04cb6b393c146ba2b53ba.png
[20210401070421954.png]: https://img-blog.csdnimg.cn/20210401070421954.png
[20210401070738845.png]: https://img-blog.csdnimg.cn/20210401070738845.png
[20210401070938251.png]: https://img-blog.csdnimg.cn/20210401070938251.png
[20210401071105973.png]: https://img-blog.csdnimg.cn/20210401071105973.png
[20210401071125318.png]: https://img-blog.csdnimg.cn/20210401071125318.png
[20210401071307919.png]: https://img-blog.csdnimg.cn/20210401071307919.png
[2021040409305610.png]: https://img-blog.csdnimg.cn/2021040409305610.png
[20210404132026860.png]: https://img-blog.csdnimg.cn/20210404132026860.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210404140441610.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2021040414055443.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210404140928790.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210404141001425.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXdhbmxl,size_16,color_FFFFFF,t_70