关于Springcloud Stream消息驱动的使用小结

红太狼 2022-10-27 15:58 162阅读 0赞

**目录**

1. Stream消息驱动概述

1.1 简介

1.2 设计思想

2. Stream案例

2.1 消息驱动之生产者

2.2 消息驱动之消费者

3. 分组消费与持久化

3.1 重复消费问题

3.2 消息持久化问题

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

# 1. Stream消息驱动概述 #

## 1.1 简介 ##

> Spring Cloud Stream is a framework for building highly scalable event-driven microservices connected with shared messaging systems.
> 
> The framework provides a flexible programming model built on already established and familiar Spring idioms and best practices, including support for persistent pub/sub semantics, consumer groups, and stateful partitions.

[Stream消息驱动][Stream]是一个构建驱动微服务的框架，应用程序通过input或者output来与Springcloud stream中binder对象交互，而Springcloud stream的binder对象负责与消息中间件交互。如此，我们只需要了解如何与Springcloud stream交互就能很方便地使用消息驱动了。（**目前仅支持RabbitMQ和Kafka**）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70][]

## 1.2 设计思想 ##

1) 标准MQ：

*  生产者/消费者之间靠消息媒介传递消息内容（**Message**）。
 *  消息必须走特定的通道（消息通道**MessageChannel**）。
 *  消息通道**MessageChannel**的子接口**SubscribableChannel**，由**MessageHandler**消息处理器所订阅。

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

*  RabbitMQ中的Exchange。
 *  Kafka中的Topic。

3) 实现应用程序与消息中间件细节之间的隔离。Stream对消息中间件的进一步封装，做到了代码层面对中间件的无感知，甚至于动态切换中间件，使得微服务开发的高度解耦，可以更多的管住自己业务流程的开发。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 1][]

# 2. Stream案例 #

以下总结以RabbitMQ为例，相关的配置可以参考：[关于Springcloud Bus的使用小结][Springcloud Bus]。

以下案例代码的[Github地址][Github]。

四个modules分别是：cloud-eureka-server7001，cloud-stream-rabbitmq-provider8801，cloud-stream-rabbitmq-consumer8802，cloud-stream-rabbitmq-consumer8803。

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

1) cloud-stream-rabbitmq-provider8801的Stream依赖。

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

2) **Binder**的配置（生产者**Output**）。

bindings: #服务的整合处理
    	output: #这个名字是一个通道的名称
    	  destination: studyExchange #表示要使用的Exchange名称定义
    	  content-type: application/json #设置消息类型，本次为json，本文要设置为“text/plain”
    	  binder: defaultRabbit #设置要绑定的消息服务的具体设置

3) 定义消息的推送管道。

@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;
        }
    }

4) 启动RabbitMQ，然后启动cloud-eureka-server7001，cloud-stream-rabbitmq-provider8801。

5) 访问多几次接口（localhost:8801/sendMessage），然后查看RabbitMQ（localhost:15672/\#/）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 3][]

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

1) cloud-stream-rabbitmq-consumer8802的依赖。

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

2) **Binder**的配置（消费者**Input**）。

bindings: #服务的整合处理
    	input: #这个名字是一个通道的名称
    	  destination: studyExchange #表示要使用的Exchange名称定义
    	  content-type: application/json #设置消息类型，本次为json，本文要设置为“text/plain”
    	  binder: defaultRabbit #设置要绑定的消息服务的具体设置

3) 定义消息的监听器。

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

4) 访问接口（localhost:8801/sendMessage），控制台即时打印监听到的消息。

![20210205001956417.png][]

5) 同样的方式创建cloud-stream-rabbitmq-consumer8803。

# 3. 分组消费与持久化 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 4][]

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

## 3.1 重复消费问题 ##

1) cloud-stream-rabbitmq-consumer8802和cloud-stream-rabbitmq-consumer8803的**group**不同。

*  8802修改YML：bindings下添加group: shenzhen。
 *  8803修改YML：bindings下添加group: guangzhou。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 5][]

访问两次接口（localhost:8801/sendMessage）。

cloud-stream-rabbitmq-provider8801：

![20210205003923253.png][]

cloud-stream-rabbitmq-consumer8802：

![20210205004016408.png][]

cloud-stream-rabbitmq-consumer8803：

![20210205004043148.png][]

2) cloud-stream-rabbitmq-consumer8802和cloud-stream-rabbitmq-consumer8803的**group相**同。

访问两次接口（localhost:8801/sendMessage）。

cloud-stream-rabbitmq-provider8801：

![20210205004317856.png][]

cloud-stream-rabbitmq-consumer8802：

![20210205004341705.png][]

cloud-stream-rabbitmq-consumer8803：

![20210205004358514.png][]

## 3.2 消息持久化问题 ##

**有指定分组的服务消息可以持久化**，即使服务中途断开后重启仍然可以取得消息，而未指定分组的服务就会丢失断开期间的MQ消息。

[Stream]: https://spring.io/projects/spring-cloud-stream
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70]: /images/20221024/7f630d68a7974e049f7b9632e0150bce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 1]: /images/20221024/9668a21c78324918ac79c06d43bbe373.png
[Springcloud Bus]: https://blog.csdn.net/BAStriver/article/details/113619364
[Github]: https://github.com/BAStriver/springcloud-test
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 2]: /images/20221024/debc2f41a4e1488189f83ab1822270ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 3]: /images/20221024/4db18ec31c964ff7acb0a80c4598c0f9.png
[20210205001956417.png]: /images/20221024/bb8f50ba095e4a05a91e8cfcef7c88eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 4]: /images/20221024/d0e1c27b4e7b4eb5be49b7609823879d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JBU3RyaXZlcg_size_16_color_FFFFFF_t_70 5]: /images/20221024/6d3d4de2326a40d59323e1f406256bd6.png
[20210205003923253.png]: /images/20221024/b23d21db1a7d4cf180682810fb4719de.png
[20210205004016408.png]: /images/20221024/fcc9531121c64c62bb1981d583beccb2.png
[20210205004043148.png]: /images/20221024/8c2e33bfec4f40459c94afc652f3e81b.png
[20210205004317856.png]: /images/20221024/2d90988fb2cf445482a2487635156299.png
[20210205004341705.png]: /images/20221024/bb9da7cab9274e39aa37d1f04fa8cc1c.png
[20210205004358514.png]: /images/20221024/9d97df2928f944d19a9c1d455ab27e8d.png