前一章咱们讲了《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. 音讯的发送接管实现
发送音讯
@Autowired
private 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})
@SpringBootApplication
public 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
七、请关注我的公众号
发表回复