1. 概述
1.1 是什么
官方定义SpringCloud Stream是一个构建消息驱动微服务的框架
应用程序通过inputs或者outputs来与Spring Cloud Stream中binder对象交互。
通过我们配置来binding( 绑定)，而SpringCloud Stream的binder对象负责与消息中间件交互。
所以，我们只需要搞清楚如何与Spring Cloud Stream交互就可以方便使用消息驱动的方式
通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。
Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。
一句话总结 ：屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型
官网
https://spring.io/projects/spring-cloud-stream#overview
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
1.2 设计思想
1.2.1 标准MQ
生产者/消费者之间靠消息(Message)媒介传递信息内容
消息必须走特定的通道 (MessageChannel)
消息通道里的消息如何被消费呢，谁负责收发谁处理
1.2.2 为什么用Cloud Stream
比方说我们用到了RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，像RabbitMQ有exchange，kafka有Topic和Partitions分区
这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个空难性的，一大堆东西都要推倒重新做，因为它跟我们的系统耦合了，这时候springcloud Stream给我们提供了一种解耦合的方式。
stream凭什么可以统一底层差异？
在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性，通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel通道，使得应用程序不需要再考虑各种不同的消息中间件实现
通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离
1.2.3 stream中的消息通信方式遵循了发布-订阅模式
Topic主题进行广播
在RabbitMQ就是Exchange, 在kafka中就是Topic
1.3 SpringCloud Stream标准流程套路
1.3.1 Binder
很方便的连接中间件，屏蔽差异
1.3.2 Channel 通道，是队列queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channel对队列进行配置
1.3.3 Source和Sink
简单的可理解为参照对象是Spring Cloud Stream自身，从Stream发布消息就是输出，接受消息就是输入。
1.4 编码API 和常用注解
2. 案例说明
工程中新建三个子模块
- cloud-stream-rabbitmq-provider8801 作为生产者进行发消息模块
- cloud-stream-rabbitmq-consumer8802 作为消息接收模块
- cloud-stream-rabbitmq-consumer8803 作为消息接收模块
3. 消息驱动之生产者
3.1 新建cloud-stream-rabbitmq-provider8801 Module
3.2 pom文件
<dependencies>
<!-- stream RabbitMQ-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!-- eureka-client-->
<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>
<!-- devtools-->
<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>
3.3 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:
instance:
prefer-ip-address: true #注册服务的时候使用服务的ip地址
lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔 (默认30秒)
lease-expiration-duration-in-seconds: 5 #如果现在超过了5秒的间隔(默认90秒)
instance-id: send-8801.com #在信息列表时显示主机名称
client:
service-url:
defaultZone: http://localhost:7001/eureka
3.4 主启动类
@SpringBootApplication
public class StreamMQMain8801 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8801.class, args);
}
}
3.5 业务类
3.5.1 发送消息接口
public interface IMessageProvider {
public String send();
}
3.5.2 发送消息接口实现类
@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;
}
}
3.5.3 Controller
@RestController
public class SendMessageController {
@Resource
private IMessageProvider messageProvider;
@GetMapping(value="/sendMessage")
public String sendMessage(){
return messageProvider.send();
}
}
3.6 测试
- 启动eureka7001
- 启动rabbitmq
- 启动cloud-stream-rabbitmq-provider8801
- 访问 http://localhost:8801/sendMessage
4. 消息驱动之消费者
4.1 新建cloud-stream-rabbitmq-consumer8802
4.2 pom文件
<dependencies>
<!-- stream RabbitMQ-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!-- eureka-client-->
<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>
<!-- devtools-->
<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>
4.3 yml文件
server:
port: 8802
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: #服务的整合处理
input: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型，本次为json, 文本则设置为"text/plain"
binder: defaultRabbit #设置要绑定的消息服务的具体设置
eureka:
instance:
prefer-ip-address: true #注册服务的时候使用服务的ip地址
lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔 (默认30秒)
lease-expiration-duration-in-seconds: 5 #如果现在超过了5秒的间隔(默认90秒)
instance-id: receive-8802.com #在信息列表时显示主机名称
client:
service-url:
defaultZone: http://localhost:7001/eureka
4.4 主启动类StreamMQMain8802
@SpringBootApplication
public class StreamMQMain8802 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8802.class, args);
}
}
4.5 业务类
package com.jd.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;
@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);
}
}
4.6 测试8801发送 8802接收消息
http://localhost:8801/sendMessage
5. 分组消费与持久化
5.1 依照8802 clone出一份cloud-stream-rabbitmq-consumer8803
5.2 启动
- 先启动RabbitMQ
- 再启动eureka7001
- 最后启动cloud-stream-rabbitmq-consumer8801 /8802/8803
运行后有两个问题
- 有重复消费问题
- 消息持久化问题
5.3 消费
5.3.1 目前是8802/8803同时都收到了，存在重复消费问题
http://localhost:8801/sendMessage
故障现象：重复消费
导致原因？
- 默认分组group是不同的，组流水号不一样，被认为不同组，可以消费
5.3.2 如何解决
自定义配置分为同一个组，解决重复消费问题
5.3.3 生产实际案例
比如在如下场景中，订单系统我们做集群部署，都会从RabbitMQ中获取订单信息，那如果一个订单同时被两个服务获取到，那么就会造成数据错误，我们得避免这种情况。这时我们就可以使用Stream中的消息分组来解决
注意在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。不同组是可以重复消费的，同一组内会发生竞争关系，只有其中一个可以消费
5.4 分组
5.4.1 原理
微服务应用放置于同一个group中，就能保证消息只会被其中一个应用消费一次。不同的组是可以消费的，同一个组内会发生竞争关系，只有其中一个可以消费。
5.4.2 8802/8803都变成了不同组，两个group不同
group: jdA、jdB
5.4.2.1 8802修改YML
bindings: #服务的整合处理
input: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型，本次为json, 文本则设置为"text/plain"
binder: defaultRabbit #设置要绑定的消息服务的具体设置
group: jdA
5.4.2.2 8803修改YML
bindings: #服务的整合处理
input: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型，本次为json, 文本则设置为"text/plain"
binder: defaultRabbit #设置要绑定的消息服务的具体设置
group: jdB
分布式微服务应用为了实现高可用和负载均衡，实际上都会部署多个实例，多数情况，生产者发送消息给某个具体微服务时只希望被消费一次，按照上面我们启动两个应用的例子，虽然它们同属一个应用，但是这个消息出现了被重复消费两次的情况。为了解决这个问题，在spring cloud Stream中提供了消费组的概念。
5.4.2.3 结论 ： 还是重复消费
5.4.3 8802/8803实现了轮询分组，每次只有一个消费者，8801模块发的消息只能被8802或8803其中一个接收到，这样就避免了重复消费
5.4.4 8802/8803都变成相同组，group两个相同
group: jdA
5.4.4.1 8802修改YML
bindings: #服务的整合处理
input: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型，本次为json, 文本则设置为"text/plain"
binder: defaultRabbit #设置要绑定的消息服务的具体设置
group: jdA
5.4.4.2 8803修改YML
bindings: #服务的整合处理
input: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型，本次为json, 文本则设置为"text/plain"
binder: defaultRabbit #设置要绑定的消息服务的具体设置
group: jdA
5.4.4.3 结论：
同一个组的多个微服务实例，每次只会有一个取到消息
这样就避免了重复消费
5.5 持久化
通过上述配置解决了重复消费问题，再看看持久化
5.5.1 停止8802/8803并去除8802的分组 jdA
5.5.2 8801 先发送4条消息到rabbitmq
5.5.3 先启动8802，无分组属性配置，后台没有打出来消息
5.5.4 再启动8803，有分组属性配置，后台打出了MQ上的消息
