Spring Cloud Stream
在微服务的开发过程中，可能会经常用到消息中间件，通过消息中间件在服务与服务之间传递消息，不管你使用的是哪款消息中间件，比如RabbitMQ、Kafka和RocketMQ，那么消息中间件和服务之间都有一点耦合性，这个耦合性就是指如果我原来使用的RabbitMQ，现在要替换为RocketMQ，那么我们的微服务都需要修改，变动会比较大，因为这两款消息中间件有一些区别，如果我们使用Spring Cloud Stream来整合我们的消息中间件，那么这样就可以降低微服务和消息中间件的耦合性，做到轻松在不同消息中间件间切换，当然Spring Cloud Stream官方只支持rabbitmq 和 kafka，spring cloud alibaba新写了一个starter可以支持RocketMQ；
按照官方的定义，Spring Cloud Stream 是一个构建消息驱动微服务的框架；
Spring Cloud Stream解决了开发人员无感知的使用消息中间件的问题，因为Spring Cloud Stream对消息中间件的进一步封装，可以做到代码层面对消息中间件的无感知，甚至于动态的切换中间件(rabbitmq切换为rocketmq或者kafka)，使得微服务开发的高度解耦，服务可以关注更多自己的业务流程；
Spring Cloud Stream 内部有几个概念：Binder 、Binding、input、output；
1、Binder: 跟外部消息中间件集成的组件，用来创建Binding，各消息中间件都有自己的 Binder 实现；
比如 Kafka 的实现 KafkaMessageChannelBinder，RabbitMQ 的实现 RabbitMessageChannelBinder 以及 RocketMQ 的实现 RocketMQMessageChannelBinder；
2、Binding: 包括 Input Binding 和 Output Binding；
Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触；
3、input
应用程序通过input（相当于消费者consumer）与Spring Cloud Stream中Binder交互，而Binder负责与消息中间件交互，因此，我们只需关注如何与Binder交互即可，而无需关注与具体消息中间件的交互。
4、Output
output（相当于生产者producer）与Spring Cloud Stream中Binder交互；
组成
说明
Binder
Binder是应用与消息中间件之间的封装，目前实现了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型(对应于Kafka的topic，RabbitMQ的exchange)，这些都可以通过配置文件来实现；
@Input
该注解标识输入通道，通过该输入通道接收消息进入应用程序
@Output
该注解标识输出通道，发布的消息将通过该通道离开应用程序
@StreamListener
监听队列，用于消费者的队列的消息接收
@EnableBinding
将信道channel和exchange、topic绑定在一起
消息生产者
1、创建SpringBoot应用31-rocket-spring-cloud-stream；
2、添加依赖：
配置文件
兼容性问题：
注意版本需要使用springboot2.2.5
消息发送：
消息接收：
可以通过调用SenderService中的方法进行发送信息，也可以通过在启动类中的Main方法中进行调用SenderService的方法进行发送信息：
在前面的案例中，我们已经实现了一个基础的 Spring Cloud Stream 消息传递处理操作，但在操作之中使用的是系统提供的 Source （output）、Sink（input），接下来我们来看一下自定义信道名称；
Apache RocketMQ在4.3.0版中已经支持分布式事务消息，这里RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示：
上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程；
1.事务消息发送及提交：
(1) 发送消息（half消息）；
(2) 服务端响应消息写入结果；
(3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）；
(4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）
2.补偿流程：
(1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”；
(2) Producer收到回查消息，检查回查消息对应的本地事务的状态；
(3) 根据本地事务状态，重新Commit或者Rollback；
其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况；
事务消息一共有三种状态：提交状态、回滚状态、中间状态；
TransactionStatus.CommitTransaction: 提交事务，代表消费者可以消费此消息；
TransactionStatus.RollbackTransaction: 回滚事务，代表消息将被删除，不能被消费；
TransactionStatus.Unknown: 中间状态，代表需要检查消息队列来确定状态；
MQ内部逻辑：
自定义信道-重写Source
自定义信道-重写Sink
消费者接收消息：
*spring.cloud.stream.rocketmq.binder.name-server*
RocketMQ NameServer 地址(老版本使用 namesrv-addr 配置项)；
Default: 127.0.0.1:9876.
*spring.cloud.stream.rocketmq.binder.access-key*
阿里云账号 AccessKey。
Default: null.
*spring.cloud.stream.rocketmq.binder.secret-key*
阿里云账号 SecretKey。
*spring.cloud.stream.rocketmq.binder.enable-msg-trace*
是否为 Producer 和 Consumer 开启消息轨迹功能
Default: true.
*spring.cloud.stream.rocketmq.binder.customized-trace-topic*
消息轨迹开启后存储的 topic 名称。
Default: RMQ_SYS_TRACE_TOPIC.
下面的这些配置是以 spring.cloud.stream.rocketmq.bindings..consumer. 为前缀的 RocketMQ Consumer 相关的配置。
*enable*
是否启用 Consumer；
默认值: true.
*tags*
Consumer 基于 TAGS 订阅，多个 tag 以 || 分割；
默认值: empty.
*sql*
Consumer 基于 SQL 订阅；
*broadcasting*
Consumer 是否是广播消费模式。如果想让所有的订阅者都能接收到消息，可以使用广播模式；
默认值: false.
*orderly*
Consumer 是否同步消费消息模式；
*delayLevelWhenNextConsume*
异步消费消息模式下消费失败重试策略：
-1,不重复，直接放入死信队列
0,broker 控制重试策略
>0,client 控制重试策略
默认值: 0.
*suspendCurrentQueueTimeMillis*
同步消费消息模式下消费失败后再次消费的时间间隔；
默认值: 1000.
下面的这些配置是以 spring.cloud.stream.rocketmq.bindings..producer. 为前缀的 RocketMQ Producer 相关的配置；
是否启用 Producer；
*group*
Producer group name；
*maxMessageSize*
消息发送的最大字节数；
默认值: 8249344.
*transactional*
是否发送事务消息；
*sync*
是否使用同步得方式发送消息；
*vipChannelEnabled*
是否在 Vip Channel 上发送消息；
*sendMessageTimeout*
发送消息的超时时间(毫秒)；
默认值: 3000.
*compressMessageBodyThreshold*
消息体压缩阀值(当消息体超过 4k 的时候会被压缩)；
默认值: 4096.
*retryTimesWhenSendFailed*
在同步发送消息的模式下，消息发送失败的重试次数；
默认值: 2.
*retryTimesWhenSendAsyncFailed*
在异步发送消息的模式下，消息发送失败的重试次数；
*retryNextServer*
消息发送失败的情况下是否重试其它的 broker；