前一章咱们讲了《SpringBoot RabbitMQ音讯队列的重试、超时、延时、死信队列》，从代码层面援用了十分多的rabbit特色代码，如：rabbitTemplate.convertAndSend()， @RabbitListener(queues = "xxx")等，都是很简略的代码看起来一切都是正当的，但隐约感觉代码受到了入侵。
业务的倒退对MQ的依赖越来越重，位置也越来越高，对它的需要也越来越多。比方程序生产，事务音讯，回溯生产等，性能方面也有更高要求。越来越多的趋势揭示咱们有更好MQ计划。
如果咱们将“MQ从Rabbit替换成Rocket”的计划提上议程，就会发放这是一个十分盛大的工程。以前好多服务都是用的有RabbitMQ的特色代码，如果要替换相当于所有服务的代码都要较大的更新，这带来的经营危险是微小的，须要十分多的开发测试资源的投入。
那回头来讲，咱们最开始应用rabbitmq的时候能不能尽量暗藏特色代码吗，为当前的降级替换保留可能性。
这个时候就须要应用Spring Cloud的子组件Spring Cloud Stream。它是一个构建音讯驱动微服务的框架，提供一套音讯订阅生产的规范为不同供应商的消息中间件进行集成。目前官网提供Kafka和RabbitMQ的集成实现，而阿里也实现对RocketMQ的集成。
一、 Spring Cloud Stream简介
Spring Cloud Stream利用由第三方的中间件组成。利用间的通信通过输出通道（input channel）和输入通道（output channel）实现。这些通道是由Spring Cloud Stream 注入的。而通道与内部的代理的连贯又是通过Binder实现的。
二、 RabbitMQ集成
1. 引入包
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency>2. 设置音讯输入输出通道
public interface Source { String OUTPUT = "myOutput"; @Output(OUTPUT) MessageChannel message();}public interface Sink { String INPUT = "myInput"; @Input(INPUT) SubscribableChannel sub1();}输入通道为音讯的发送方，输出通道为音讯的接管方
myOutput，myInput为通道名，后续通过配置文件进行个性配置，切记两个通道的绑定最好是离开定义，不然有可能产生莫名谬误
3. 音讯个性配置
spring cloud: stream: bindings: myOutput: destination: login-user myInput: # 通道名，对应代码中的生产监听组 destination: login-user # exchange group: logined-member # 生产组 rabbit: bindings: myOutput: producer: routing-key-expression: headers.routingKey # 发送端路由key delayed-exchange: true # 开启延时队列 myInput: consumer: binding-routing-key: login.user.succeed # 生产监听路由表达式 delayed-exchange: true # 开启延时队列 auto-bind-dlq: true # 绑定死信队列 republish-to-dlq: true # 重投到死信队列并带有报错信息1) destination音讯的主题名
在Rabbit中用来定义exchange以及成为queue的一部分
2) group生产组
- 没有定义生产组时，如果启动多实例则一个音讯同时都生产
- 定义了生产组后，多实例共用一个queue，负载生产。从图能够看出queue名为
destination.group组成
- binding-routing-key：生产路由监听表达式
- delayed-exchange： 开启延时队列
- auto-bind-dlq：开启死信队列
- republish-to-dlq：此设置能够让死信音讯带报错信息
4. 音讯的发送接管实现
发送音讯
@Autowiredprivate Source source;@GetMapping("/")public void sendSucceed() { source.message().send(MessageBuilder.withPayload("Hello World...") .setHeader("routingKey", "login.user.succeed") .setHeader("version", "1.0") .setHeader("x-delay", 5000) .build());}这里能够为音讯设置不同header，以事实不同的性能，这部分每种MQ有不同的个性，须要视状况而定
接管音讯
@StreamListener(value = Sink.MY_INPUT_1, condition = "headers['version']=='1.0'")public void receiveSucceed_v1(@Payload String message) { String msg = "StreamReceiver v1: " + message; log.error(msg);}5. 绑定音讯通道
@EnableBinding(value = {Source.class, Sink.class})@SpringBootApplicationpublic class RabbitApplication { public static void main(String[] args) { SpringApplication.run(RabbitApplication.class, args); }}实现这5步就能够失常发送接管音讯了，你会发现除了引入不同的包和音讯个性配置外，其它的代码都是形象代码，没有任何rabbitmq的特色代码
三、 RocketMQ集成
依据RabbitMQ的相干代码，只须要批改引入包和特片配置就能够替换成RocketMQ了（一些个性性能除外）
1. 引入包
<dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-stream-rocketmq</artifactId></dependency>2. 音讯特色配置
spring cloud: stream: bindings: myOutput: destination: login-user content-type: application/json myInput: # 通道名，对应代码中的生产监听组 destination: login-user # exchange group: logined-member # 消费者组, 同组负载生产 rocketmq: binder: name-server: 127.0.0.1:9876四、 Kafka集成
1. 引入包
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-kafka</artifactId></dependency>2. 音讯特色配置
spring cloud: stream: bindings: myOutput: destination: login-user content-type: application/json myInput: # 通道名，对应代码中的生产监听组 destination: login-user # exchange group: logined-member # 消费者组, 同组负载生产 kafka: binder: brokers: localhost:9092 #Kafka的消息中间件服务器 auto-create-topics: true五、 总结
由下面三个简略的例子能够看出，Spring Cloud Stream对音讯订阅和生产做了高度形象，用一套代码实现多种消息中间件的反对。同时它也能够非常简单的实现多种消息中间件的混用，大大扩大了消息中间件的玩法。
这里也倡议如果没有非凡的特色场景须要实现的话，举荐应用Spring Cloud Stream组件来实现音讯的订阅与生产，对中间件进行高度接耦。
六、源代码
文中代码因为篇幅起因有肯定省略并不是残缺逻辑，如有趣味请Fork源代码
https://gitee.com/hypier/barry-cloud/tree/master/cloud-stream