前一章我们讲了《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`消费组
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
七、请关注我的公众号
评论