Spring Cloud Stream详解

刺骨的言语ヽ痛彻心扉 2024-04-08 11:26 67阅读 0赞

#### 目录 ####

*   *  一、概述简介
     *   *  1.1. cloud Stream是什么
         *  1.2. 设计思想
         *  1.3. 标准流程
         *  1.4. 注解
     *  二、基于注解代码练习
     *   *  2.1. 消息驱动之生产者
         *  2.2. 消息驱动之消费者
         *  2.3. 目前存在的问题
         *  2.4. 分组解决重复消费问题
         *  2.5. 消息持久化
     *  三、函数式编程练习

本篇文章所涉及到的demo练习 使用的`cloud 2021.0.3+ springboot2.6.8`

### 一、概述简介 ###

官网：[https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/][https_docs.spring.io_spring-cloud-stream_docs_current_reference_html]

官网概述：[https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html\#spring-cloud-stream-preface-notable-deprecations][https_docs.spring.io_spring-cloud-stream_docs_current_reference_html_spring-cloud-stream.html_spring-cloud-stream-preface-notable-deprecations]

#### 1.1. cloud Stream是什么 ####

官方定义：Spring Cloud Stream是一个用于构建 与 `共享消息系统` 连接的高度可扩展的`事件驱动微服务`。

**目前主流的消息框架有：**

*  ActiveMQ
 *  RabbitMQ
 *  RocketMQ
 *  Kafka

> 假设公司业务项目用了RabbitMQ，而大数据项目用了Kafka。这时候就会出现有两个消息框架，相对于程序员来说其实并不友好，还得两个都掌握，正常对于一个程序员来说熟练一个消息框架都不错了，何况还搞了两个，并且两个维护起来也不好维护。

> RabbitMQ和Kafka是两个不同的框架，两个消息模型上也存在着差异，并且代码上用法也不一样。Spring Cloud Stream就是不再关注具体MQ的细节，可以在不改代码的基础上，来完成Rabbit和Kafka两个不同的消息中间件的切换（这里的切换指的是原本用的RabbitMQ，但是用着用着发现kafka比较符合，所以想要换框架）。

**总结成一句话：屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型**

也就是基于Stream可以和如下消息中间件来整合使用！

![在这里插入图片描述][92645724164c40318aaf38dc17db5ca3.png]

#### 1.2. 设计思想 ####

**常规的MQ设计如下：**  
![在这里插入图片描述][42275e22a3d744599e8de4a44c7d4ac2.png]

*  Message：生产者/消费者之间靠消息媒介传递信息内容
 *  MessageChannel：消息必须走特定的通道
 *  队列：假如发消息会先发到消息队列当中
 *  消息队列的消息如何被消费呢：订阅的人可以进行消费

**cloud Stream设计如下：**

通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离。

> 在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性，通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件细节之间的隔离。Stream对消息中间件的进一步封装，可以做到代码层面对中间件的无感知，甚至于动态的切换中间件(rabbitmq切换为kafka)，使得微服务开发的高度解耦，服务可以关注更多自己的业务流程

注意：左图是官网的架构图  
![在这里插入图片描述][8ab5014afa8c4b9e9a317a7e04bc88e6.png]

Binder可以生成Binding，Binding用来绑定消息容器的生产者和消费者，它有两种类型，INPUT和OUTPUT，INPUT对应于消费者，OUTPUT对应于生产者。

![在这里插入图片描述][3bc2cf275fd8495692a284deeeff3a5e.png]

> stream为了屏蔽差异，抽象出来了一个Binder层。而spring官方提供了这两个框架的实现。

假如想要通过stream连接RabbitMQ就使用：

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

假如想要通过stream连接Kafka就使用：

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

Stream中的消息通信方式遵循了`发布-订阅模式`，Topic主题进行广播，在RabbitMQ就是`Exchange`，在Kakfa中就是`Topic`。

#### 1.3. 标准流程 ####

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjEwMTgzOQ_size_16_color_FFFFFF_t_60][]

*  **Binder：** 很方便的连接中间件,屏蔽差异
 *  **Channel：** 通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channe对队列进行配置
 *  **Source（源：发送者）和Sink（水槽：接受者）：** 简单的可理解为参照对象是Spring Cloud Stream自身，从Stream发布消息就是输出，接受消息就是输入。

#### 1.4. 注解 ####

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjEwMTgzOQ_size_16_color_FFFFFF_t_60 1][]

注解完全是基于官方给的模型而定的！通过stream使用消息中间件也是非常简单的，直接使用以下注解就可以使用。

![在这里插入图片描述][124f9d7ea7c94a669e4229888fc57946.png]

> 注意：注解依然是能用的，但是官方明确表示注解已经被弃用，弃用并不是不能用，而是用了会画横杠不建议用。但是功能是没有问题的，低版本的cloud是没有被弃用的。针对于注解和函数式编程两种我都会进行使用。

> 题外话：学技术永远是这样，技术一直在不断的更新迭代，真正学习一个技术并不是要掌握编码使用，而是要掌握他到底是什么，能干什么，要去深入理解他，对于编码，我认为其实不是很重要。就算你今天掌握了官方最新用法，回头人家又改写法了。

![在这里插入图片描述][6086e7efd3414f06abec6a2f73b25f8c.png]  
![在这里插入图片描述][62ee4cab6bdb416cbf8df7391ce00fde.png]

### 二、基于注解代码练习 ###

生产者就是消息发送者，消费者就是消息接受者。这里我就不用kafka了，我直接用的是RabbitMQ。

windows下安装RabbitMQ：[https://blog.csdn.net/weixin\_43888891/article/details/126514021][https_blog.csdn.net_weixin_43888891_article_details_126514021]

#### 2.1. 消息驱动之生产者 ####

**1.创建项目（可以是聚合可以是普通springboot项目）  
2.添加pom**

> 因为是和RabbitMQ整合，所以就是引入的spring-cloud-starter-stream-rabbit启动器

<properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <springboot.version>2.6.8</springboot.version>
        <springcloud.version>2021.0.3</springcloud.version>
    </properties>
    
    <dependencyManagement>
        <dependencies>
            <dependency>
               <groupId>org.springframework.boot</groupId>
               <artifactId>spring-boot-dependencies</artifactId>
               <version>${springboot.version}</version>
               <type>pom</type>
               <scope>import</scope>
           </dependency>
            <dependency>
               <groupId>org.springframework.cloud</groupId>
               <artifactId>spring-cloud-dependencies</artifactId>
               <version>${springcloud.version}</version>
               <type>pom</type>
               <scope>import</scope>
           </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
    </dependencies>

**3.添加application配置**

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 # 设置要绑定的消息服务的具体设置

**4.添加接口**

public interface IMessageProvider {
        
        public String send();
    }

**5.添加实现类**

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 javax.annotation.Resource;
    import java.util.UUID;
    
    // 可以理解为是一个消息的发送管道的定义
    @EnableBinding(Source.class)
    public class MessageProviderImpl implements IMessageProvider {
        
    
        // 消息的发送管道
        @Resource
        private MessageChannel output;
    
        @Override
        public String send() {
        
            String serial = UUID.randomUUID().toString();
            // 创建并发送消息
            this.output.send(MessageBuilder.withPayload(serial).build());
            System.out.println("***serial: " + serial);
            return serial;
        }
    }

**6.添加controller控制器**

@RestController
    public class SendMessageController {
        
    
        @Autowired
        private IMessageProvider iMessageProvider;
    
        @GetMapping("send")
        public String send() {
        
            return iMessageProvider.send();
        }
    }

**7.测试**

（1）首先要保证RabbitMQ是可以访问的：[http://localhost:15672][http_localhost_15672]

![在这里插入图片描述][1e4894a3057a4d8db8fc2e434fe6c2fb.png]

（2）启动项目访问：[http://localhost:8801/send][http_localhost_8801_send]

下图波峰代表发送消息成功

![在这里插入图片描述][16a8c76b4dff49c2b32fd92a479950bc.png]

启动后会创建交换机，名称就是application.yml当中的destination属性设置的

![在这里插入图片描述][63d4310a2e2e44d8951665737deb6143.png]

**注意：停止服务后并没有删除交换机！！！**

#### 2.2. 消息驱动之消费者 ####

**1.创建项目  
2.添加pom（pom和发送者依赖一模一样）**  
**3.添加application配置**

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 # 设置要绑定的消息服务的具体设置

**4.添加监听（消费者只负责接受消息）**

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;
    
    @Component
    @EnableBinding(Sink.class)
    public class ReceiveMessageListener {
        
        @Value("${server.port}")
        private String serverPort;
    
        @StreamListener(Sink.INPUT)
        public void input(Message<String> message) {
        
            System.out.println("消费者1号，------->接收到的消息：" + message.getPayload() + "\t port: " + serverPort);
        }
    }

**5.测试**

（1）启动RabbitMQ  
（2）启动发送消息端服务  
（3）启动消费者服务，启动后会发现，他自动会向这个交换机当中添加一个队列。

![在这里插入图片描述][f4f48c30459e43c5ac28526866284c97.png]  
发送消息：[http://localhost:8801/send][http_localhost_8801_send]  
接受消息：

![在这里插入图片描述][07874c065187407a8140ae94f37483b4.png]

![在这里插入图片描述][7b0520c367f9457cb29df7fe9cc9190a.png]

**注意：当停止服务后消息队列会被自动删除！！！**

#### 2.3. 目前存在的问题 ####

**1.依照8802, clone出来一份运行8803，主要用来演示多个消费者的场景**  
**2.启动8801生产者**  
**3.启动8802消费者**  
**4.启动8803消费者**

> 当三个服务都启动后通过RabbitMQ界面会发现，一个交换机绑定了两个队列

![在这里插入图片描述][23e72031571040ef9135ffac682741ab.png]

**运行后会发现存在两个问题：**

1.  有重复消费问题
2.  消息持久化问题

**（1）重复消费问题：**

发送消息后两个消费者都收到了消息：[http://localhost:8801/send][http_localhost_8801_send]

![在这里插入图片描述][3f9d0d846389491bb657f16f2fbc3779.png]

![在这里插入图片描述][d5a17cc00d814ea1961f6c949a619892.png]

> 比如在如下场景中，订单系统我们做集群部署，都会从RabbitMQ中获取订单信息，那如果一个订单同时被两个服务获取到，那么就会造成数据错误，我们得避免这种情况。这时我们就可以使用Stream中的消息分组来解决

![在这里插入图片描述][3dc92b3c6b204b908ced0af8daf3813a.png]

> 注意在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)，同一组内会发生竞争关系，只有其中一个可以消费。

**（2）消息持久化问题：**

当生产者发送消息的时候，消费者恰好宕机了，但是过一会消费者恢复了，但是消息却没收到。那也就是意味着消息队列是临时消息队列。针对于这一点，大家也可以测试一下，加深一下印象。

#### 2.4. 分组解决重复消费问题 ####

**原理：** 微服务应用放置于同一个group中,就能够保证消息只会被其中一个应用消费一次。`同一个组内会发生竞争关系,只有其中一个可以消费。`

接下来直接调整两个消费者为同一个组：添加如下配置

![在这里插入图片描述][bd60828dccca4f53a242a9734775b4ef.png]

**当两个消费者都设置好后启动，会发现一个问题：** `实际上分到一个组对于RabbitMQ来说就是两个消费者监听了一个队列。`一个队列那也就意味着，当队列收到一条消息，哪个消费者谁先消费就是谁的，消费完队列里面就没有了，也就是只有一个消费者能消费到消息！

> 注意：假如不设置group属性的时候，默认是启动一个消费者，就会创建一个消费队列，启动多个服务就会创建多个队列。stream默认使用的是RabbitMQ的topic交换机。当发送者向这个交换机发送消息的时候，两个队列就都会接收到。关于RabbitMQ相关知识本篇不记录，后续会专门写RabbitMQ相关文章。

![在这里插入图片描述][5ed3c2ef3a574c51979bf5be81975917.png]

> 最终测试：8802/8803实现了轮询分组,每次只有一个消费者8801模块的发的消息只能被8802或8803其中一个接收到,这样避免了重复消费。

#### 2.5. 消息持久化 ####

当三个项目都启动着的时候，现在我们要做几件事：

1.  停止8802和8803并去除掉8802的分组`group: gxs`，8803不去分组信息，停止掉项目的时候会发现消息队列并没有删除，说明一旦设置分组信息，消息队列就不再是临时队列。

![在这里插入图片描述][f0ba5fab4fc343718423f9c3fd4b7853.png]

1.  8801发送4条消息
2.  启动8802然后消息并没有打印，没有收到消息（注意8802是去掉分组信息的）
3.  再启动8803,有分组属性配置,后台打出来了MQ上的消息

> 原因就是：当两个项目都停止的时候，队列并未删除，而8803还绑定了这个队列，所以他就算宕机了，又重启了，依然可以收到消息。而8802没有设置分组信息，他再启动后系统会给他创建一个临时队列，自然而然收不到之前的消息了。

### 三、函数式编程练习 ###

官网介绍：[https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html\#spring\_cloud\_function][https_docs.spring.io_spring-cloud-stream_docs_current_reference_html_spring-cloud-stream.html_spring_cloud_function]

由于文章篇幅过长，函数式编程将在下一篇文章来记录！

[https_docs.spring.io_spring-cloud-stream_docs_current_reference_html]: https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/
[https_docs.spring.io_spring-cloud-stream_docs_current_reference_html_spring-cloud-stream.html_spring-cloud-stream-preface-notable-deprecations]: https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-preface-notable-deprecations
[92645724164c40318aaf38dc17db5ca3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/aa91bc3d076c46a182220f2594854d63.png
[42275e22a3d744599e8de4a44c7d4ac2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/13287ade87a94c3a8d090d2850573114.png
[8ab5014afa8c4b9e9a317a7e04bc88e6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/1470bae36b27451ebebb4977740ed819.png
[3bc2cf275fd8495692a284deeeff3a5e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/2a5650e5e051445da14db8256c96e18b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjEwMTgzOQ_size_16_color_FFFFFF_t_60]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/d74db381fb924ddcb5117a33353f593d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjEwMTgzOQ_size_16_color_FFFFFF_t_60 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/8ca186e14fa74b2ca05a0674029754ae.png
[124f9d7ea7c94a669e4229888fc57946.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/6f9734f45b5c4bb7bb6889d69989c562.png
[6086e7efd3414f06abec6a2f73b25f8c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/91bea1d8e0104d79a410924a04cebc4c.png
[62ee4cab6bdb416cbf8df7391ce00fde.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/bc2414f2149145d198cd8b363e9c7831.png
[https_blog.csdn.net_weixin_43888891_article_details_126514021]: https://blog.csdn.net/weixin_43888891/article/details/126514021
[http_localhost_15672]: http://localhost:15672
[1e4894a3057a4d8db8fc2e434fe6c2fb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/09b553e7d3b14243af14f330051a0ac3.png
[http_localhost_8801_send]: http://localhost:8801/send
[16a8c76b4dff49c2b32fd92a479950bc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/422dcb2fb1344fbe9c76e39ba5eedb6e.png
[63d4310a2e2e44d8951665737deb6143.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/49d6603639d942d3a7ffd3789154be0e.png
[f4f48c30459e43c5ac28526866284c97.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/7a299b4c6a4a49cda6af60be09dfb1e9.png
[07874c065187407a8140ae94f37483b4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/d1cc32facf1d42179e941ce4fdbc94de.png
[7b0520c367f9457cb29df7fe9cc9190a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/538e96d4329e4bfda647c35f414d321c.png
[23e72031571040ef9135ffac682741ab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/898e9a594ebd418eba821991835bef08.png
[3f9d0d846389491bb657f16f2fbc3779.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/d61dd7ea522944afb28b4f3fb46bd4f9.png
[d5a17cc00d814ea1961f6c949a619892.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/6df260cb37b54179ab6ea1c4be9ea83b.png
[3dc92b3c6b204b908ced0af8daf3813a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/fef57e1061d94042a07de28c2f3f3c3a.png
[bd60828dccca4f53a242a9734775b4ef.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/b178870996e64d41a87661211d2b325c.png
[5ed3c2ef3a574c51979bf5be81975917.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/a8112707ad92499b9e6d29255cbf6097.png
[f0ba5fab4fc343718423f9c3fd4b7853.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/c6ba79c37e094487a5922b18413e4aba.png
[https_docs.spring.io_spring-cloud-stream_docs_current_reference_html_spring-cloud-stream.html_spring_cloud_function]: https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#spring_cloud_function