写点什么

使用 Spring Cloud Stream 玩转 RabbitMQ,RocketMQ 和 Kafka

发布于: 2020 年 09 月 27 日
使用Spring Cloud Stream玩转RabbitMQ,RocketMQ和Kafka

前一章我们讲了《SpringBoot RabbitMQ消息队列的重试、超时、延时、死信队列》,从代码层面引用了非常多的 rabbit 特征代码,如:rabbitTemplate.convertAndSend() @RabbitListener(queues = "xxx")等,都是很简单的代码看起来一切都是合理的,但隐约感觉代码遭到了入侵。


业务的发展对 MQ 的依赖越来越重,地位也越来越高,对它的需求也越来越多。比如顺序消费,事务消息,回溯消费等,性能方面也有更高要求。越来越多的趋势提醒我们有更好 MQ 方案。


假如我们将“MQ 从 Rabbit 替换成 Rocket”的方案提上议程,就会发放这是一个非常浩大的工程。以前好多服务都是用的有RabbitMQ的特征代码,如果要替换相当于所有服务的代码都要较大的更新,这带来的运营风险是巨大的,需要非常多的开发测试资源的投入。


那回头来讲,我们最开始使用 rabbitmq 的时候能不能尽量隐藏特征代码吗,为以后的升级替换保留可能性。


这个时候就需要使用Spring Cloud的子组件Spring Cloud Stream。它是一个构建消息驱动微服务的框架,提供一套消息订阅消费的标准为不同供应商的消息中间件进行集成。目前官方提供KafkaRabbitMQ的集成实现,而阿里也实现对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();}
复制代码


输出通道为消息的发送方,输入通道为消息的接收方


myOutputmyInput为通道名,后续通过配置文件进行特性配置,切记两个通道的绑定最好是分开定义,不然有可能产生莫名错误


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


七、请关注我的公众号


用户头像

还未添加个人签名 2019.07.10 加入

还未添加个人简介

评论

发布
暂无评论
使用Spring Cloud Stream玩转RabbitMQ,RocketMQ和Kafka