MQ协助——Spring Cloud Stream

骑猪看日落 2023-10-15 16:46 2阅读 0赞

## 一、概念 ##

Spring Cloud Stream 是一个构建消息驱动微服务的框架。

![fe8279e22dac489c826484aeb5937906.png][]

提供多种消息中间件的配置。理解为 RedisTemplate 、RestTemplate 简单概况：封装方法、简化开发复杂度，程序员更专心业务处理。

![8b2191f0dd0447899eb90425905a5feb.png][]

## 二、核心概念 ##

1.  Binder（绑定器）：用于将**应用程序与消息中间件进行绑定**，以便发送和接收消息。Spring Cloud Stream 支持多种消息中间件，通过配置 Binder 可以方便地切换底层消息中间件的实现。
2.  Channel（通道）：是消息在**应用程序内部流动的渠道**。Spring Cloud Stream 使用了发布订阅模式，消息通过通道进行传递。通道可以分为输入通道（input channel）和输出通道（output channel）。
3.  Binder 绑定注解：用于将应用程序的输入和输出**绑定到具体的通道**上。
4.  消息转换器：用于将**消息的数据格式进行转换**，以适应不同的业务需求。

![286b303c991845ccb3d4a24f152471f4.png][]

> 以下是 Spring Cloud Stream 的一些主要模块：
> 
> 1.  Spring Cloud Stream Core：核心模块，提供了对 Binder 和 Channel 的支持。
> 2.  Spring Cloud Stream RabbitMQ：提供与 RabbitMQ 消息中间件的集成。
> 3.  Spring Cloud Stream Kafka：提供与 Apache Kafka 消息中间件的集成。
> 4.  Spring Cloud Stream Binder：提供与其他消息中间件的集成，例如 ActiveMQ、Redis 等。

例子：RabbitMQ

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

### 分区支持 ###

分区支持指的是将消息发送到指定的分区中。分区是消息队列的一种划分方式，在分布式环境中用于实现负载均衡和并行处理。

多用在生产者端，每个分区都有一个标识符，可以是数字、字符串或其他形式的标识

一个应用程序多个实例时，数据推送到哪个/哪台服务器接收。topic代理被视为多分区结构

Spring Cloud Stream 提供一个通用的抽象，统一分区处理。

故分区可以用于自带分区的代理（如kafka）或者不带分区的代理（如rabbiemq）

**注意：分区在有状态处理中是一个很重要的概念，其重要性体现在性能和一致性上，要确保所有相关数据被一并处理**

**举例：应用程序多个实例名为s1、s2、s3....保证结果都为s1处理，其他实例不处理**

## 三、编程模型 ##

Spring Cloud Stream提供了一些预定义的注解，用于绑定输入和输出channels，以及如何监听channels。

> **1.@EnableBinding触发绑定。将应用程序与消息通道进行绑定。**
> 
> 可以实现将消息通道注入到应用程序中，从而方便地进行消息的发送和接收操作。
> 
>     @Import(...)
>     @Configuration
>     @EnableIntegration
>     public @interface EnableBinding {
>         ...
>         Class<?>[] value() default {};
>     }
> 
> @EnableBinding注解本身就包含@Configuration注解，会触发Spring Cloud Stream 基本配置
> 
> #### @StreamListener 定义在方法中，被修饰的方法注册为消息中间件上数据流的事件监听器，注解中属性值对应了监听的消息通道名 ####
> 
> 举例：
> 
>     @EnableBinding(MyProcessor.class)
>     public class MyApplication {
>     
>         public static void main(String[] args) {
>             SpringApplication.run(MyApplication.class, args);
>         }
>     
>         @Autowired
>         private MyProcessor myProcessor;
>     //获取消息INPUT为消息通道名称    
>         @StreamListener(MyProcessor.INPUT)
>         public void handleMessage(String message) {
>             System.out.println("Received message: " + message);
>             // 处理接收到的消息
>         }
>     //发送消息到消息通道中
>         public void sendMessage(String message) {
>             myProcessor.output().send(MessageBuilder.withPayload(message).build());
>         }
>     }
> 
> **2.@Input 与 @Output**
> 
> 一个Spring Cloud Stream应用可以有任意数目的input和output通道
> 
> 使用 @Input 和 @Output 注解来声明 INPUT 和 OUTPUT 消息通道
> 
> **Spring Cloud Stream提供了三个开箱即用的预定义接口。Source、Sink、Processor**
> 
>     public interface Source {
>     //作用于单个输出通道的应用
>       String OUTPUT = "output";
>     
>       @Output(Source.OUTPUT)
>       MessageChannel output();
>     
>     }
>     public interface Sink {
>     //作用于单个输入通道的应用
>       String INPUT = "input";
>     
>       @Input(Sink.INPUT)
>       SubscribableChannel input();
>     
>     }
>     public interface Processor extends Source, Sink {
>     // 作用于单个应用同时包含输入和输出通道
>     }
> 
> 举例：MyProcessor 参考 Processor手动拷贝一个
> 
>     public interface MyProcessor {
>     
>         String INPUT = "inputChannel";
>         String OUTPUT = "outputChannel";
>     
>         @Input(INPUT)
>         SubscribableChannel input();
>     
>         @Output(OUTPUT)
>         MessageChannel output();
>     }
> 
> **1.中例子改为原始版，应用程序自己接受、并推送消息**
> 
>     @SpringBootApplication
>     @EnableBinding(Processor.class)
>     public class MyLoggerServiceApplication {
>         public static void main(String[] args) {
>             SpringApplication.run(MyLoggerServiceApplication.class, args);
>         }
>     
>         @StreamListener(Processor.INPUT)//接收
>         @SendTo(Processor.OUTPUT)//推送
>         public LogMessage enrichLogMessage(LogMessage log) {
>             return new LogMessage(String.format("[1]: %s", log.getMessage()));
>         }
>     }
> 
> 解释：
> 
> **@EnableBinding声明了这个应用程序绑定了2个通道【Processor继承输入、输出即Sink、Source】**：INPUT和OUTPUT。这2个通道是在接口Processor中定义的（Spring Cloud Stream默认设置）。所有通道都是配置在一个具体的消息中间件或绑定器中。  
> **@StreamListener(Processor.INPUT)表明这里在input中提取消息，并且处理。**  
> ***@SendTo(Processor.OUTPUT)表明在output中返回消息。 跟1中初版对比下看看啊***

## 官网：[Spring Cloud Stream][] ##

[参考： 官方文档中文版！Spring Cloud Stream 快速入门 - 邴越 - 博客园 (cnblogs.com)][_Spring Cloud Stream _ - _ - _ _cnblogs.com]

[fe8279e22dac489c826484aeb5937906.png]: https://img-blog.csdnimg.cn/fe8279e22dac489c826484aeb5937906.png
[8b2191f0dd0447899eb90425905a5feb.png]: https://img-blog.csdnimg.cn/8b2191f0dd0447899eb90425905a5feb.png
[286b303c991845ccb3d4a24f152471f4.png]: https://img-blog.csdnimg.cn/286b303c991845ccb3d4a24f152471f4.png
[Spring Cloud Stream]: https://spring.io/projects/spring-cloud-stream
[_Spring Cloud Stream _ - _ - _ _cnblogs.com]: https://www.cnblogs.com/binyue/p/12222198.html