spring-cloud-stream

左手的ㄟ右手 2024-02-18 09:15 6阅读 0赞

## 系列文章目录 ##

[第一章 Java线程池技术应用][Java]  
[第二章 CountDownLatch和Semaphone的应用][CountDownLatch_Semaphone]  
[第三章 Spring Cloud 简介][Spring Cloud]  
[第四章 Spring Cloud Netflix 之 Eureka][Spring Cloud Netflix _ Eureka]  
[第五章 Spring Cloud Netflix 之 Ribbon][Spring Cloud Netflix _ Ribbon]  
[第六章 Spring Cloud 之 OpenFeign][Spring Cloud _ OpenFeign]  
[第七章 Spring Cloud 之 GateWay][Spring Cloud _ GateWay]  
[第八章 Spring Cloud Netflix 之 Hystrix][Spring Cloud Netflix _ Hystrix]  
[第九章 代码管理gitlab 使用][_gitlab]  
[第十章 SpringCloud Alibaba 之 Nacos discovery][SpringCloud Alibaba _ Nacos discovery]  
[第十一章 SpringCloud Alibaba 之 Nacos Config][SpringCloud Alibaba _ Nacos discovery]  
[第十二章 Spring Cloud Alibaba 之 Sentinel][Spring Cloud Alibaba _ Sentinel]  
[第十三章 JWT][JWT]  
[第十四章 RabbitMQ应用][RabbitMQ]  
[第十五章 RabbitMQ 延迟队列][RabbitMQ 1]  
[第十六章 spring-cloud-stream][spring-cloud-stream]

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

--------------------

### 文章目录  系列文章目录  @\[TOC\](文章目录)  前言  1、stream设计思想  2、编码常用的注解  3、编码步骤  3.1、添加依赖  3.2、修改配置文件  3.3、生产  3.4、消费  3.5、延迟队列  3.5.1、修改配置文件  3.5.2、生产端  3.5.2、消息确认机制 消费端  总结 ###

## 前言 ##

https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit  
官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。应用程序通过inputs或者outputs来与Spring Cloud Stream中binder对象交互。通过我们配置来binding(绑定)，而Spring Cloud Stream的binder对象负责与消息中间件交互。

SpringCloud stream通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。

Stream让我们不再关注具体MQ的细节我们只需要用一种适配绑定的方式，自动的给我们在各种MQ内切换，总的来说Stream能够屏蔽底层消息中间件的差异、降低切换成本，是统一消息的编程模型。

## 1、stream设计思想 ##

![在这里插入图片描述][af7bf946d0834cf18f2ffecc3f8d1112.png]  
![在这里插入图片描述][95e3a12f3f074b6eb291d5781dac7b63.png]

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

## 2、编码常用的注解 ##

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

<table> 
 <thead> 
  <tr> 
   <th align="left">组成</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">Middleware</td> 
   <td align="left">中间件，目前只支持RabbitMQ和Kafka</td> 
  </tr> 
  <tr> 
   <td align="left">Binder</td> 
   <td align="left">Binder是应用与消息中间件之间的封装，目前实现了Kafka和RabbitMQ的Binder，通过BInder可以很方便的连接中间件，可以动态的改变消息类型（对应于Kafka的topic，RabbitMQ的exchange）,这些都可以通过配置文件来实现。</td> 
  </tr> 
  <tr> 
   <td align="left">@Input</td> 
   <td align="left">注解标识输入通道，通过该输入通道接收到的消息进入应用程序</td> 
  </tr> 
  <tr> 
   <td align="left">@Output</td> 
   <td align="left">注解标识输出通道，发布的消息将通过该通道离开应用程序</td> 
  </tr> 
  <tr> 
   <td align="left">@StreamListener</td> 
   <td align="left">监听队列，用于消费者的队列的消息接收</td> 
  </tr> 
  <tr> 
   <td align="left">@EnableBinding</td> 
   <td align="left">指信道channel和exchange绑定在一起</td> 
  </tr> 
 </tbody> 
</table>

## 3、编码步骤 ##

### 3.1、添加依赖 ###

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

### 3.2、修改配置文件 ###

server:
      port: 8088
    
    spring:
      cloud:
        stream:
          binders: #需要绑定的rabbitmq的服务信息
            defaultRabbit:  #定义的名称，用于bidding整合
              type: rabbit  #消息组件类型
              environment:  #配置rabbimq连接环境
                spring:
                  rabbitmq:
                    host: localhost   #rabbitmq 服务器的地址
                    port: 5672           #rabbitmq 服务器端口
                    username: tiger       #rabbitmq 用户名
                    password: tiger       #rabbitmq 密码
                    virtual-host: tiger_vh  #虚拟路径
          bindings:        #服务的整合处理
            saveOrderOutput:    #这个是消息通道的名称 --->保存订单输出通道
              destination: exchange-saveOrder     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。
              content-type: application/json      #设置消息的类型，本次为json
              default-binder: defaultRabbit
              group: saveOrderGroup               #分组
            saveOrderInput: #生产者绑定，这个是消息通道的名称---> 保存订单输入通道
              destination: exchange-saveOrder     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。
              content-type: application/json      #设置消息的类型，本次为json
              default-binder: defaultRabbit
              group: saveOrderGroup               #分组

### 3.3、生产 ###

/**
     * 订单消息输出通道处理器
     */
    @Component
    public interface OrderOutputChannelProcesor {
        
        @Output("saveOrderOutput")
        MessageChannel saveOrderOutput();
    }

@Slf4j
    @EnableBinding(OrderOutputChannelProcesor.class)
    public class OrderMessageProducer {
        
        @Autowired
        @Output("saveOrderOutput")
        private MessageChannel messageChannel;
    
        public void sentMsg(UserInfo userInfo){
        
            messageChannel.send(MessageBuilder.withPayload(userInfo).build());
            log.info("消息发送成功：" + userInfo);
        }
    }

### 3.4、消费 ###

/**
     * 订单消息输入通道处理器
     */
    @Component
    public interface OrderInputChannelProcesor {
        
        @Input("saveOrderInput")
        SubscribableChannel saveOrderInput();
    }

@Slf4j
    @EnableBinding(OrderInputChannelProcesor.class)
    public class OrderMessageConsumer {
        
        @StreamListener("saveOrderInput")
        public void receiveMsg(Message<UserInfo> userInfoMessage){
        
            log.info("接收消息成功：" + userInfoMessage.getPayload());
        }
    }

### 3.5、延迟队列 ###

安装延迟队列插件：  
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.11.1/rabbitmq\_delayed\_message\_exchange-3.11.1.ez  
下载解压，到plugins目录，执行以下的命令：

> rabbitmq-plugins enable rabbitmq\_delayed\_message\_exchange

#### 3.5.1、修改配置文件 ####

server:
      port: 8088
    
    spring:
      cloud:
        stream:
          binders: #需要绑定的rabbitmq的服务信息
            defaultRabbit:  #定义的名称，用于bidding整合
              type: rabbit  #消息组件类型
              environment:  #配置rabbimq连接环境
                spring:
                  rabbitmq:
                    host: localhost   #rabbitmq 服务器的地址
                    port: 5672           #rabbitmq 服务器端口
                    username: tiger       #rabbitmq 用户名
                    password: tiger       #rabbitmq 密码
                    virtual-host: tiger_vh  #虚拟路径
          bindings:        #服务的整合处理
            saveOrderOutput:    #这个是消息通道的名称 --->保存订单输出通道
              destination: exchange-saveOrder-delay     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。
              content-type: application/json      #设置消息的类型，本次为json
              default-binder: defaultRabbit
              group: saveOrderGroup               #分组
            saveOrderInput: #生产者绑定，这个是消息通道的名称---> 保存订单输入通道
              destination: exchange-saveOrder-delay     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。
              content-type: application/json      #设置消息的类型，本次为json
              default-binder: defaultRabbit
              group: saveOrderGroup               #分组
          rabbit:
            bindings: #服务的整合处理
              saveOrderOutput:    #这个是消息通道的名称 --->保存订单输出通道
                producer:
                  delayed-exchange: true
              saveOrderInput:
                consumer:
                  delayed-exchange: true

#### 3.5.2、生产端 ####

@Slf4j
    @EnableBinding(OrderOutputChannelProcesor.class)
    public class OrderMessageProducer {
        
        @Autowired
        @Output("saveOrderOutput")
        private MessageChannel messageChannel;
    
        public void sentMsg(UserInfo userInfo){
        
            messageChannel.send(MessageBuilder.withPayload(userInfo).setHeader("x-delay", 5000).build());
            log.info("消息发送成功：" + userInfo);
        }
    }

#### 3.5.2、消息确认机制 消费端 ####

rabbit:
      bindings: #服务的整合处理
        saveOrderInput:
          consumer:
            acknowledge-mode: MANUAL #手动确认

@StreamListener("saveOrderInput")
    public void receiveMsg(Message<UserInfo> userInfoMessage){
        
    
        log.info("接收消息成功：" + userInfoMessage.getPayload());
        Channel channel = (Channel) userInfoMessage.getHeaders().get(AmqpHeaders.CHANNEL);
        Long delieverTag = (Long) userInfoMessage.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        /*
         * deliveryTag：Channel的消息投递的唯一标识符。
         * multiple：是否否定应答多条消息。如果设置为true，则否定应答带指定deliveryTag的消息及该deliveryTag之前的多条消息；
         * 如果设置为false，则仅否定应答带指定deliveryTag的单条消息。
         * requeue：被否定应答的消息是否重入队列。如果设置为true，则消息重入队列；
         * 如果设置为false，则消息被丢弃或发送到死信Exchange。
         */
        try {
        
            channel.basicAck(delieverTag,true);
        } catch (IOException e) {
        
            e.printStackTrace();
        }
    }

定义交换机类型为direct

rabbit:
      bindings: #服务的整合处理
        saveOrderInput:
          consumer:
            bindingRoutingKey: orderRoutingKey
            bindQueue: true
            exchangeType: direct
        saveOrderOutput:
          producer:
            routingKeyExpression: orderRoutingKey
            exchangeType: direct

--------------------

## 总结 ##

spring-cloud-stream目前支持RabbitMQ和Kafka，与spring-cloud无缝集成，非常方便。

[Java]: https://blog.csdn.net/s445320/article/details/132870714
[CountDownLatch_Semaphone]: https://blog.csdn.net/s445320/article/details/132912126
[Spring Cloud]: https://blog.csdn.net/s445320/article/details/132939152
[Spring Cloud Netflix _ Eureka]: https://blog.csdn.net/s445320/article/details/133024912
[Spring Cloud Netflix _ Ribbon]: https://blog.csdn.net/s445320/article/details/133156575
[Spring Cloud _ OpenFeign]: https://blog.csdn.net/s445320/article/details/130624259
[Spring Cloud _ GateWay]: https://blog.csdn.net/s445320/article/details/133612365
[Spring Cloud Netflix _ Hystrix]: https://blog.csdn.net/s445320/article/details/133759729
[_gitlab]: https://blog.csdn.net/s445320/article/details/133842594
[SpringCloud Alibaba _ Nacos discovery]: https://blog.csdn.net/s445320/article/details/133977189
[Spring Cloud Alibaba _ Sentinel]: https://blog.csdn.net/s445320/article/details/134149339
[JWT]: https://blog.csdn.net/s445320/article/details/134197122
[RabbitMQ]: https://blog.csdn.net/s445320/article/details/134228181
[RabbitMQ 1]: https://blog.csdn.net/s445320/article/details/130823776
[spring-cloud-stream]: https://blog.csdn.net/s445320/article/details/134360837
[8692ff151c5c4afdaf8f38ad8911ceef.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/85249e8585d04a29ae7f469197c41585.png
[af7bf946d0834cf18f2ffecc3f8d1112.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/68d976c9d013447fbe6ad672c5c77e0c.png
[95e3a12f3f074b6eb291d5781dac7b63.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/6a6da55bb94349518a429d81e3cb7148.png
[6b92d0c04a1a4904a5d0001842b385ab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/d5e7339d96204a88bb0fcaed71cd77b3.png