我的家乡网站设计模板,51zwd一起做网店,网站建设保障机制,可视化网站开发平台本文将对rocktmq-spring-boot的设计实现做一个简单的介绍#xff0c;读者可以通过本文了解将RocketMQ Client端集成为spring-boot-starter框架的开发细节#xff0c;然后通过一个简单的示例来一步一步的讲解如何使用这个spring-boot-starter工具包来配置#xff0c;发送和消…本文将对rocktmq-spring-boot的设计实现做一个简单的介绍读者可以通过本文了解将RocketMQ Client端集成为spring-boot-starter框架的开发细节然后通过一个简单的示例来一步一步的讲解如何使用这个spring-boot-starter工具包来配置发送和消费RocketMQ消息。 作者简介辽天阿里巴巴技术专家Apache RocketMQ 内核控拥有多年分布式系统研发经验对Microservice、Messaging和Storage等领域有深刻理解 目前专注 RocketMQ 内核优化以及 Messaging 生态建设。 通过本文您将了解到
Spring的消息框架介绍rocketmq-spring-boot具体实现使用示例插播一条广告本周六下午Apache RocketMQ 开发者沙龙将来到杭州欢迎大家到现场活动详情请点击“阅读原文”。 前言
上世纪90年代末随着Java EE(Enterprise Edition)的出现特别是Enterprise Java Beans的使用需要复杂的描述符配置和死板复杂的代码实现增加了广大开发者的学习曲线和开发成本由此基于简单的XML配置和普通Java对象(Plain Old Java Objects)的Spring技术应运而生依赖注入(Dependency Injection), 控制反转(Inversion of Control)和面向切面编程(AOP)的技术更加敏捷地解决了传统Java企业及版本的不足。
随着Spring的持续演进基于注解(Annotation)的配置逐渐取代了XML文件配置 2014年4月1日Spring Boot 1.0.0正式发布它基于“约定大于配置”Convention over configuration)这一理念来快速地开发、测试、运行和部署Spring应用并能通过简单地与各种启动器(如 spring-boot-web-starter)结合让应用直接以命令行的方式运行不需再部署到独立容器中。这种简便直接快速构建和开发应用的过程可以使用约定的配置并且简化部署受到越来越多的开发者的欢迎。
Apache RocketMQ是业界知名的分布式消息和流处理中间件简单地理解它由Broker服务器和客户端两部分组成 其中客户端一个是消息发布者客户端(Producer)它负责向Broker服务器发送消息 另外一个是消息的消费者客户端(Consumer)多个消费者可以组成一个消费组来订阅和拉取消费Broker服务器上存储的消息。 为了利用Spring Boot的快速开发和让用户能够更灵活地使用RocketMQ消息客户端Apache RocketMQ社区推出了spring-boot-starter实现。随着分布式事务消息功能在RocketMQ 4.3.0版本的发布近期升级了相关的spring-boot代码通过注解方式支持分布式事务的回查和事务消息的发送。
本文将对当前的设计实现做一个简单的介绍读者可以通过本文了解将RocketMQ Client端集成为spring-boot-starter框架的开发细节然后通过一个简单的示例来一步一步的讲解如何使用这个spring-boot-starter工具包来配置发送和消费RocketMQ消息。
Spring 中的消息框架
顺便在这里讨论一下在Spring中关于消息的两个主要的框架即Spring Messaging和Spring Cloud Stream。它们都能够与Spring Boot整合并提供了一些参考的实现。和所有的实现框架一样消息框架的目的是实现轻量级的消息驱动的微服务可以有效地简化开发人员对消息中间件的使用复杂度让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。
2.1 Spring Messaging
Spring Messaging是Spring Framework 4中添加的模块是Spring与消息系统集成的一个扩展性的支持。它实现了从基于JmsTemplate的简单的使用JMS接口到异步接收消息的一整套完整的基础架构Spring AMQP提供了该协议所要求的类似的功能集。 在与Spring Boot的集成后它拥有了自动配置能力能够在测试和运行时与相应的消息传递系统进行集成。
单纯对于客户端而言Spring Messaging提供了一套抽象的API或者说是约定的标准对消息发送端和消息接收端的模式进行规定不同的消息中间件提供商可以在这个模式下提供自己的Spring实现在消息发送端需要实现的是一个XXXTemplate形式的Java Bean结合Spring Boot的自动化配置选项提供多个不同的发送消息方法在消息的消费端是一个XXXMessageListener接口实现方式通常会使用一个注解来声明一个消息驱动的POJO提供回调方法来监听和消费消息这个接口同样可以使用Spring Boot的自动化选项和一些定制化的属性。
如果有兴趣深入的了解Spring Messaging及针对不同的消息产品的使用推荐阅读这个文件。参考Spring Messaging的既有实现RocketMQ的spring-boot-starter中遵循了相关的设计模式并结合RocketMQ自身的功能特点提供了相应的API(如顺序异步和事务半消息等)。
2.2 Spring Cloud Stream
Spring Cloud Stream结合了Spring Integration的注解和功能它的应用模型如下 Spring Cloud Stream框架中提供一个独立的应用内核它通过输入(Input)和输出(Output)通道与外部世界进行通信消息源端(Source)通过输入通道发送消息消费目标端(Sink)通过监听输出通道来获取消费的消息。这些通道通过专用的Binder实现与外部代理连接。开发人员的代码只需要针对应用内核提供的固定的接口和注解方式进行编程而不需要关心运行时具体的Binder绑定的消息中间件。在运行时Spring Cloud Stream能够自动探测并使用在classpath下找到的Binder。
这样开发人员可以轻松地在相同的代码中使用不同类型的中间件仅仅需要在构建时包含进不同的Binder。在更加复杂的使用场景中也可以在应用中打包多个Binder并让它自己选择Binder甚至在运行时为不同的通道使用不同的Binder。
Binder抽象使得Spring Cloud Stream应用可以灵活的连接到中间件加之Spring Cloud Stream使用利用了Spring Boot的灵活配置配置能力这样的配置可以通过外部配置的属性和Spring Boo支持的任何形式来提供包括应用启动参数、环境变量和application.yml或者application.properties文件部署人员可以在运行时动态选择通道连接destination例如Kafka的topic或者RabbitMQ的exchange。
Binder SPI的方式来让消息中间件产品使用可扩展的API来编写相应的Binder并集成到Spring Cloud Steam环境目前RocketMQ还没有提供相关的Binder我们计划在下一步将完善这一功能也希望社区里有这方面经验的同学积极尝试贡献PR或建议。
spring-boot-starter的实现
在开始的时候我们已经知道spring boot starter构造的启动器对于使用者是非常方便的使用者只要在pom.xml引入starter的依赖定义相应的编译运行和部署功能就全部自动引入。因此常用的开源组件都会为Spring的用户提供一个spring-boot-starter封装给开发者让开发者非常方便集成和使用这里我们详细的介绍一下RocketMQ客户端的starter实现过程。
3.1. spring-boot-starter的实现步骤
对于一个spring-boot-starter实现需要包含如下几个部分
在pom.xml的定义
定义最终要生成的starter组件信息
groupIdorg.apache.rocketmq/groupId
artifactIdspring-boot-starter-rocketmq/artifactId
version1.0.0-SNAPSHOT/version
定义依赖包
它分为两个部分: A、Spring自身的依赖包 B、RocketMQ的依赖包
dependencies!-- spring-boot-start internal depdencies --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter/artifactId/dependency dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependency!-- rocketmq dependencies --dependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion${rocketmq-version}/version/dependency
/dependencies dependencyManagementdependencies!-- spring-boot-start parent depdency definition -- dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion${spring.boot.version}/versiontypepom/typescopeimport/scope/dependency/dependencies
/dependencyManagement配置文件类
定义应用属性配置文件类RocketMQProperties,这个Bean定义一组默认的属性值。用户在使用最终的starter时可以根据这个类定义的属性来修改取值当然不是直接修改这个类的配置而是spring-boot应用中对应的配置文件src/main/resources/application.properties.
定义自动加载类
定义 src/resources/META-INF/spring.factories文件中的自动加载类 其目的是让spring boot更具文中中所指定的自动化配置类来自动初始化相关的BeanComponent或Service它的内容如下
org.springframework.boot.autoconfigure.EnableAutoConfiguration\
org.apache.rocketmq.spring.starter.RocketMQAutoConfiguration
在RocketMQAutoConfiguration类的具体实现中定义开放给用户直接使用的Bean对象. 包括
RocketMQProperties 加载应用属性配置文件的处理类RocketMQTemplate 发送端用户发送消息的发送模板类ListenerContainerConfiguration 容器Bean负责发现和注册消费端消费实现接口类这个类要求由RocketMQMessageListener注解标注实现RocketMQListener泛化接口。
最后具体的RocketMQ相关的封装 在发送端producer和消费端(consumer)客户端分别进行封装在当前的实现版本提供了对Spring Messaging接口的兼容方式。
3.2. 消息发送端实现
普通发送端
发送端的代码封装在RocketMQTemplate POJO中下图是发送端的相关代码的调用关系图 为了与Spring Messaging的发送模板兼容在RocketMQTemplate集成了AbstractMessageSendingTemplate抽象类来支持相关的消息转换和发送方法这些方法最终会代理给doSend()方法doSend()以及RocoketMQ所特有的一些方法如异步单向和顺序等方法直接添加到RoketMQTempalte中这些方法直接代理调用到RocketMQ的Producer API来进行消息的发送。
事务消息发送端
对于事务消息的处理在消息发送端进行了部分的扩展参考下图的调用关系类图
RocketMQTemplate里加入了一个发送事务消息的方法sendMessageInTransaction(), 并且最终这个方法会代理到RocketMQ的TransactionProducer进行调用在这个Producer上会注册其关联的TransactionListener实现类以便在发送消息后能够对TransactionListener里的方法实现进行调用。
3.3. 消息消费端实现 在消费端Spring-Boot应用启动后会扫描所有包含RocketMQMessageListener注解的类(这些类需要集成RocketMQListener接口并实现onMessage()方法)这个Listener会一对一的被放置到DefaultRocketMQListenerContainer容器对象中容器对象会根据消费的方式(并发或顺序)将RocketMQListener封装到具体的RocketMQ内部的并发或者顺序接口实现。在容器中创建RocketMQ Consumer对象启动并监听定制的Topic消息如果有消费消息则回调到Listener的onMessage()方法。
使用示例
上面的一章介绍了RocketMQ在spring-boot-starter方式的实现这里通过一个最简单的消息发送和消费的例子来介绍如何使这个rocketmq-spring-boot-starter。
4.1 RocketMQ服务端的准备
启动NameServer和Broker
要验证RocketMQ的Spring-Boot客户端首先要确保RocketMQ服务正确的下载并启动。可以参考RocketMQ主站的快速开始来进行操作。确保启动NameServer和Broker已经正确启动。
创建实例中所需要的Topics
在执行启动命令的目录下执行下面的命令行操作
bash bin/mqadmin updateTopic -c DefaultCluster -t string-topic
4.2. 编译rocketmq-spring-boot-starter
目前的spring-boot-starter依赖还没有提交的Maven的中心库用户使用前需要自行下载git源码然后执行mvn clean install 安装到本地仓库。
git clone https://github.com/apache/rocketmq-externals.git
cd rocketmq-spring-boot-starter
mvn clean install
4.3. 编写客户端代码
用户如果使用它需要在消息的发布和消费客户端的maven配置文件pom.xml中添加如下的依赖
properties spring-boot-starter-rocketmq-version1.0.0-SNAPSHOT/spring-boot-starter-rocketmq-version
/propertiesdependencygroupIdorg.apache.rocketmq/groupIdartifactIdspring-boot-starter-rocketmq/artifactIdversion${spring-boot-starter-rocketmq-version}/version
/dependency
属性spring-boot-starter-rocketmq-version的取值为1.0.0-SNAPSHOT 这与上一步骤中执行安装到本地仓库的版本一致。
消息发送端的代码
发送端的配置文件application.properties
# 定义name-server地址
spring.rocketmq.name-serverlocalhost:9876
# 定义发布者组名
spring.rocketmq.producer.groupmy-group1
# 定义要发送的topic
spring.rocketmq.topicstring-topic
发送端的Java代码
import org.apache.rocketmq.spring.starter.core.RocketMQTemplate;
...SpringBootApplication
public class ProducerApplication implements CommandLineRunner {// 声明并引用RocketMQTemplateResourceprivate RocketMQTemplate rocketMQTemplate;// 使用application.properties里定义的topic属性Value(${spring.rocketmq.springTopic})private String springTopic;public static void main(String[] args){SpringApplication.run(ProducerApplication.class, args);}public void run(String... args) throws Exception {// 以同步的方式发送字符串消息给指定的topicSendResult sendResult rocketMQTemplate.syncSend(springTopic, Hello, World!);// 打印发送结果信息System.out.printf(string-topic syncSend1 sendResult%s %n, sendResult);}
}
消息消费端代码
消费端的配置文件application.properties
# 定义name-server地址
spring.rocketmq.name-serverlocalhost:9876
# 定义发布者组名
spring.rocketmq.consumer.groupmy-customer-group1
# 定义要发送的topic
spring.rocketmq.topicstring-topic
消费端的Java代码
SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}
}// 声明消费消息的类并在注解中指定相关的消费信息
Service
RocketMQMessageListener(topic ${spring.rocketmq.topic}, consumerGroup ${spring.rocketmq.consumer.group})
class StringConsumer implements RocketMQListenerString {Overridepublic void onMessage(String message) {System.out.printf(------- StringConsumer received: %s %f, message);}
} 原文链接 本文为云栖社区原创内容未经允许不得转载。
