Java工程师的进阶之路1 - SpringCloud简介与工程构建2 - 服务注册中心：Eureka、Zookeeper、Consul3 - 服务负载均衡：Ribbon4 - 服务接口调用：OpenFeign5 - 服务的断路器：Hystrix6 - 服务的路由器：GateWay7 - 服务配置中心：SpringCloud Config8 - 服务消息驱动：SpringCloud Stream9 - 服务链路跟踪：SpringCloud Sleuth
目录
十三、SpringCloud Stream
13.1、简介13.2、标准流程13.3、编码API及常用注解13.4、准备工作13.5、发送端13.6、消费端13.7、消息重复消费13.8、group解决消息重复消费13.9、消息持久化
十三、SpringCloud Stream 13.1、简介
1、什么是Spring Cloud Stream？
官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。
屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型
应用程序通过inputs或者 outputs 来与Spring Cloud Stream中binder对象交互。
通过我们配置来binding(绑定)，而Spring Cloud Stream 的binder对象负责与消息中间件交互。所以，我们只需要搞清楚如何与Spring Cloud Stream交互就可以方便使用消息驱动的方式。
通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。
Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。
目前仅支持RabbitMQ、 Kafka
2、Stream的设计思想
传统MQ执行流程
生产者/消费者之间靠消息媒介传递信息内容消息必须走特定的通道 - 消息通道 Message Channel消息通道里的消息如何被消费呢，谁负责收发处理 - 消息通道MessageChannel的子接口SubscribableChannel，由MessageHandler消息处理器所订阅。
3、为什么用Cloud Stream？
不同中间件存在差异
比方说我们用到了RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，像RabbitMQ有exchange，kafka有Topic和Partitions分区。
这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都要重新推倒重新做，因为它跟我们的系统耦合了
但Spring Cloud Stream给我们提供了—种解耦合的方式
屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型
4、Stream凭什么可以统一底层差异？
在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel通道，使得应用程序不需要再考虑各种不同的消息中间件实现。
通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离
Binder：
INPUT对应于消费者OUTPUT对应于生产者
Stream中的消息通信方式遵循了发布-订阅模式
Topic主题进行广播
在RabbitMQ就是Exchange在Kakfa中就是Topic
13.2、标准流程
简化图
详细内部流程
Binder - 很方便的连接中间件，屏蔽差异（无需考虑使用哪种MQ，之间用Binder作为桥梁）Channel - 通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channel对队列进行配置。Source和Sink - 简单的可理解为参照对象是Spring Cloud Stream自身，从Stream发布消息就是输出，接受消息就是输入。
13.3、编码API及常用注解
| 组成 | 说明 |
|---|---|
| Middleware | 中间件，目前只支持RabbitMQ和Kafka |
| Binder | Binder是应用与消息中间件之间的封装，目前实行了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型(对应于Kafka的topic,RabbitMQ的exchange)，这些都可以通过配置文件来实现 |
| @Input | 注解标识输入通道，通过该输乎通道接收到的消息进入应用程序 |
| @Output | 注解标识输出通道，发布的消息将通过该通道离开应用程序 |
| @StreamListener | 监听队列，用于消费者的队列的消息接收 |
| @EnableBinding | 指信道channel和exchange绑定在一起 |
13.4、准备工作
1、准备RabbitMQ环境（见上文 12.7目录）本次教程将MQ环境配置在阿里云的Docker中
2、工程中新建三个子模块
cloud-stream-rabbitmq-provider8801，作为生产者进行发消息模块cloud-stream-rabbitmq-consumer8802，作为消息接收模块cloud-stream-rabbitmq-consumer8803，作为消息接收模块
13.5、发送端
1、新建Module：cloud-stream-rabbitmq-provider8801
2、POM
org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-actuator org.springframework.cloud spring-cloud-starter-netflix-eureka-client org.springframework.cloud spring-cloud-starter-stream-rabbit org.springframework.boot spring-boot-devtools runtime true org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test
3、YML
server:
port: 8801
spring:
application:
name: cloud-stream-provider
rabbitmq:
host: 120.76.55.55
port: 5672
username: guest
password: guest
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息；
defaultRabbit: # 表示定义的名称，用于于binding整合
type: rabbit # 消息组件类型
bindings: # 服务的整合处理
output: # 这个名字是一个通道的名称
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
instance-id: send-8801.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址
4、主启动类
@SpringBootApplication
public class StreamMQMain8801 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8801.class,args);
}
}
5、业务类
发送消息接口
public interface IMessageProvider {
public String send();
}
接口实现类
@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;
}
}
6、控制层
@RestController
public class SendMessageController {
@Resource
private IMessageProvider messageProvider;
@GetMapping(value = "/sendMessage")
public String sendMessage() {
return messageProvider.send();
}
}
7、测试
启动 7001eureka、RabpitMQ、8801
访问 - http://localhost:15672/
访问 - http://localhost:8801/sendMessage
后台将打印serial: UUID字符串
13.6、消费端
1、新建Module - cloud-stream-rabbitmq-consumer8802
2、POM
org.springframework.boot spring-boot-starter-web org.springframework.cloud spring-cloud-starter-netflix-eureka-client org.springframework.cloud spring-cloud-starter-stream-rabbit org.springframework.boot spring-boot-starter-actuator org.springframework.boot spring-boot-devtools runtime true org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test
3、YML
server:
port: 8802
spring:
application:
name: cloud-stream-consumer
rabbitmq:
host: 120.76.55.55
port: 5672
username: guest
password: guest
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息；
defaultRabbit: # 表示定义的名称，用于于binding整合
type: rabbit # 消息组件类型
bindings: # 服务的整合处理
input: # 这个名字是一个通道的名称
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
instance-id: send-8802.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址
4、主启动类
@SpringBootApplication
public class StreamMQMain8802 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8802.class,args);
}
}
5、业务类
@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {
@Value("${server.port}")
private String serverPort;
@StreamListener(Sink.INPUT)
public void input(Message message) {
System.out.println("消费者1号,----->接受到的消息: "+message.getPayload()+"t port: "+serverPort);
}
}
6、测试
启动EurekaMain7001、StreamMQMain8801、启动StreamMQMain8802
访问 - http://localhost:8801/sendMessage8801发送消息后8802接收到消息
流程图：
13.7、消息重复消费
1、按照8802消费者作为模板新建8803模块
2、启动注册中心7001、发送端8801、接收端8802/8803
3、访问 - http://localhost:8801/sendMessage8801发送消息后8802/8003都接收到消息
存在重复消费及消息持久化的问题
生产实际案例：
比如在如下场景中，订单系统我们做集群部署，都会从RabbitMQ中获取订单信息，那如果一个订单同时被两个服务获取到，那么就会造成数据错误，我们得避免这种情况。这时我们就可以使用Stream中的消息分组来解决。
注意在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)。
13.8、group解决消息重复消费
思路：
微服务应用放置于同一个group中，就能够保证消息只会被其中一个应用消费一次。不同的组是可以重复消费的，同一个组内会发生竞争关系，只有其中一个可以消费。
优化：
将8002和8003变成同一个组 Laptoy_Group
优化 8002/8003 YML 添加如下
spring:
cloud:
stream:
bindings: # 服务的整合处理
input: # 这个名字是一个通道的名称
group: Laptoy_Group
访问 - http://localhost:8801/sendMessage 8002/8003每次只有其中一台获取到消息，默认为轮询策略
13.9、消息持久化
测试流程：
- 去除8803的分组配置、8802的配置保留启动 - 配置中心7001、发送端8001发送端8801 - 发送4条消息到RabbitMq启动 - 8002，后台打出来了MQ上的消息。(消息持久化体现)启动 - 8803，后台没有打出来消息
总结：消息持久化需要接收端开启分组配置才能实现
