文章目录
- 前言
- 技术积累
- Spring Cloud Stream3.0新特性
- RocketMq简介
- 实战演示
- 引入Maven依赖
- 增加application配置
- 消息生产者
- 消息消费者
前言
相信很多同学用使用过rocketmq消息中间件,且大多情况下是使用原生的rocketmq-spring-boot-starter 进行集成然后创建一个rocketMQTemplate发送的生产者和@RocketMQMessageListener监听的消费者。今天我们就不按常理出牌,使用Spring Cloud Stream来进行整合RocketMq。如果我们有在一个项目中需要引入多个MQ的需求,用Spring Cloud Stream简直不要太好。当然,我们是直接使用Spring Cloud Stream3.0,不再像之前低版本那样需要引入通道类进行指定,3.0版本可用直接字配置文件进行粘接绑定信道,简直不要太爽。
技术积累
Spring Cloud Stream3.0新特性
Spring Cloud Stream 3.0 引入了一些新特性,包括对新版本Spring Boot和Spring Cloud的支持,以及对反序列化错误处理的改进。
以下是一些主要的新特性:
支持Spring Boot 2.x和Spring Cloud 2020.0.x。
改进了消息中间件的错误处理,提供了更好的异常传播和提供了更多的配置选项来自定义错误处理。
提供了对函数式编程模型的支持。
提供了对Kafka消息传递保证的配置选项。
提供了对消息转换器的支持,可以在发送和接收消息之前进行自定义转换。
RocketMq简介
RocketMQ是Apache基金会下的一个开源分布式消息中间件,设计用于云原生环境,支持高吞吐量和流处理,广泛应用于金融级稳定性场景。 它具备以下核心特性:
云原生:RocketMQ设计为与云和容器技术(如Kubernetes)友好,支持无限弹性的扩缩。
高吞吐:能够保证万亿级别的吞吐量,满足微服务与大数据场景的需求。
流处理:提供轻量、高扩展、高性能和丰富功能的流计算引擎。
金融级稳定性:广泛用于交易核心链路,确保系统的稳定运行。
架构极简:采用零外部依赖的Shared-nothing架构,简化系统设计和维护。
生态友好:无缝对接微服务、实时计算、数据湖等周边生态,便于集成和使用。
支持多种消息类型:包括普通消息、顺序消息、事务消息、批量消息、定时(延时)消息、消息回溯等,满足不同业务场景需求。
易用性与灵活性:提供多种发送与消费模式,丰富的客户端支持,以及易于运维与管理的工具和界面。
实战演示
今天的重点不要RocketMq的使用,而是Spring Cloud Stream3.0如何整合RocketMq。以下是一个简单的整合DEMO,仅供学习使用,如果需要应用与生产环境需要增加一些额外的方案。比如死信或者消费失败重试机制等等。
引入Maven依赖
这里需要注意SpringBoot与SpriingCloud版本对应,SpringCloud版本与RocketMq Starter版本对应
<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.12.RELEASE</version><relativePath/> <!-- lookup parent from repository -->
</parent>
<properties><java.version>8</java.version><spring-cloud.version>Hoxton.SR12</spring-cloud.version>
</properties>
</dependencies><!--rocketmq--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-stream-rocketmq</artifactId><version>2.2.2.RELEASE</version><exclusions><exclusion><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId></exclusion><exclusion><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-acl</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.7.1</version></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-acl</artifactId><version>4.7.1</version></dependency></dependencies>
<dependencyManagement><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>${spring-cloud.version}</version><type>pom</type><scope>import</scope></dependency></dependencies>
</dependencyManagement>
增加application配置
testChannel可以作为输入输出信道
spring:cloud:stream:rocketmq:binder:name-server: 127.0.0.1:9876binders:my-rocketmq:type: rocketmqfunction:definition: testChannelbindings:testChannel-in-0:binder: my-rocketmqdestination: test-rocket-topicgroup: test-rocket-groupcontent-type: text/plain# 设置spring cloud stream次数1,表示禁用,异常情况下只消费一次消息consumer:max-attempts: 1testChannel-out-0:binder: my-rocketmqdestination: test-rocket-topiccontent-type: text/plain
消息生产者
直接可以用过StreamBridge 进行手动发送
@RestController
@RequestMapping("/base")
public class BaseController {@Resourceprivate StreamBridge streamBridge;//@Resourceprivate MqChannel mqChannel;@GetMapping("/send")public Boolean sendMessage(String msg) {boolean send = streamBridge.send("testChannel-out-0", MessageBuilder.withPayload("rocket测试:" + msg).build());return true;}
}
消息消费者
直接监听testChannel通道,默认监听testChannel-input-0信道
/*** RocketChannel* @author senfel* @version 1.0* @date 2024/7/23 12:20*/
@Configuration
public class RocketChannel {/*** testChannel 消费者* @author senfel* @date 2024/7/23 12:26* @return java.util.function.Consumer<java.lang.String>*/@Beanpublic Consumer<Message<String>> testChannel(){return message -> {System.out.println("接收到消息Payload:" + message.getPayload());System.out.println("接收到消息Header:" + message.getHeaders());};}
}
测试用例