- 当前位置：首页 >域名 >如何在优雅地Spring中实现消息的发送和消费
如何在优雅地Spring中实现消息的发送和消费
发布时间：2025-11-05 15:22:33 来源：码上建站 作者：系统运维
前言
上世纪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)结合，现消息让应用直接以命令行的发送费方式运行，不需再部署到独立容器中。这种简便直接快速构建和开发应用的过程，可以使用约定的配置并且简化部署，受到越来越多的云南idc服务商开发者的欢迎。
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
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实现需要包含如下几个部分：
1. 在pom.xml的定义
定义最终要生成的starter组件信息 <groupId>org.apache.rocketmq</groupId> <artifactId>spring-boot-starter-rocketmq</artifactId> <version>1.0.0-SNAPSHOT</version> 定义依赖包，它分为两个部分: A、Spring自身的依赖包； B、RocketMQ的依赖包
<dependencies> <!-- spring-boot-start internal depdencies --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <!-- rocketmq dependencies --> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>${rocketmq-version}</version> </dependency> </dependencies> <dependencyManagement> <dependencies> <!-- spring-boot-start parent depdency definition --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>${spring.boot.version}</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement>2. 配置文件类
定义应用属性配置文件类RocketMQProperties,这个Bean定义一组默认的属性值。用户在使用最终的starter时，可以根据这个类定义的属性来修改取值，当然不是直接修改这个类的配置，而是spring-boot应用中对应的配置文件：src/main/resources/application.properties.
3. 定义自动加载类
定义 src/resources/META-INF/spring.factories文件中的自动加载类， 其目的是让spring boot更具文中中所指定的自动化配置类来自动初始化相关的Bean，Component或Service，它的内容如下：
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ org.apache.rocketmq.spring.starter.RocketMQAutoConfiguration在RocketMQAutoConfiguration类的具体实现中，定义开放给用户直接使用的Bean对象. 包括：
RocketMQProperties 加载应用属性配置文件的处理类； RocketMQTemplate 发送端用户发送消息的发送模板类； ListenerContainerConfiguration 容器Bean负责发现和注册消费端消费实现接口类，这个类要求：由@RocketMQMessageListener注解标注；实现RocketMQListener泛化接口。4. 最后具体的RocketMQ相关的封装
在发送端（producer）和消费端(consumer)客户端分别进行封装，在当前的实现版本提供了对Spring Messaging接口的兼容方式。
3.2. 消息发送端实现
1. 普通发送端
发送端的代码封装在RocketMQTemplate POJO中，下图是发送端的相关代码的调用关系图：
为了与Spring Messaging的发送模板兼容，在RocketMQTemplate集成了AbstractMessageSendingTemplate抽象类，来支持相关的消息转换和发送方法，这些方法最终会代理给doSend()方法；doSend()以及RocoketMQ所特有的一些方法如异步，单向和顺序等方法直接添加到RoketMQTempalte中，这些方法直接代理调用到RocketMQ的Producer API来进行消息的发送。
2. 事务消息发送端
对于事务消息的处理，在消息发送端进行了部分的扩展，参考下图的调用关系类图：
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已经正确启动。
2. 创建实例中所需要的Topics
在执行启动命令的目录下执行下面的命令行操作
bash bin/mqadmin updateTopic -c DefaultCluster -t string-topic4.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 install4.3. 编写客户端代码
用户如果使用它，需要在消息的发布和消费客户端的maven配置文件pom.xml中添加如下的依赖：
<properties> <spring-boot-starter-rocketmq-version>1.0.0-SNAPSHOT</spring-boot-starter-rocketmq-version> </properties> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>spring-boot-starter-rocketmq</artifactId> <version>${spring-boot-starter-rocketmq-version}</version> </dependency>属性spring-boot-starter-rocketmq-version的取值为：1.0.0-SNAPSHOT， 这与上一步骤中执行安装到本地仓库的版本一致。
消息发送端的代码发送端的配置文件application.properties
# 定义name-server地址 spring.rocketmq.name-server=localhost:9876 # 定义发布者组名 spring.rocketmq.producer.group=my-group1 # 定义要发送的topic spring.rocketmq.topic=string-topic发送端的Java代码
import org.apache.rocketmq.spring.starter.core.RocketMQTemplate; ... @SpringBootApplication public class ProducerApplication implements CommandLineRunner { // 声明并引用RocketMQTemplate @Resource private 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 { // 以同步的方式发送字符串消息给指定的topic SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!"); // 打印发送结果信息 System.out.printf("string-topic syncSend1 sendResult=%s %n", sendResult); } }2. 消息消费端代码
消费端的配置文件application.properties
# 定义name-server地址 spring.rocketmq.name-server=localhost:9876 # 定义发布者组名 spring.rocketmq.consumer.group=my-customer-group1 # 定义要发送的topic spring.rocketmq.topic=string-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 RocketMQListener{ @Override public void onMessage(String message) { System.out.printf("------- StringConsumer received: %s %f", message); } }这里只是简单的介绍了使用spring-boot来编写最基本的消息发送和接收的代码，如果需要了解更多的调用方式，如: 异步发送，对象消息体，指定tag标签以及指定事务消息，请参看github的说明文档和详细的代码。我们后续还会对这些高级功能进行陆续的介绍。
- 一、使用命令设置ubuntu的ip地址 1.修改配置文件blacklist.conf禁用IPV6： 2.在文档最后添加 blacklist ipv6，然后查看修改结果： 3.设置IP（设置网卡eth0的IP地址和子网掩码） 4.设置网关 5.设置DNS 修改/etc/resolv.conf，在其中加入nameserver DNS的地址1 和 nameserver DNS的地址2 完成。 6.重启网络服务（若不行，请重启ubuntu：sudo reboot）： 7.查看当前IP： 二、直接修改系统配置文件 ubuntu的网络配置文件是：/etc/network/interfaces 并用下面的行来替换有关eth0的行： 用下面的命令使网络设置生效： 也可以在命令行下直接输入下面的命令来获取地址 2. 为网卡配置静态IP地址 并用下面的行来替换有关eth0的行： 将上面的ip地址等信息换成你自己就可以了.用下面的命令使网络设置生效： 3. 设定第二个IP地址(虚拟IP地址) 在该文件中添加如下的行： 根据你的情况填上所有诸如address,netmask,network,broadcast和gateways等信息： 使用下面的命令来设置当前主机的主机名称： 系统启动时,它会从/etc/hostname来读取主机的名称。 5. 配置DNS 6.手动重启网络服务： 返回结果如下： *Reconfiguring network interfaces… [OK]
- 盘点前端问题，你知道几个？
- Python 告诉你，你为什么是单身汪
- Go 语言新提案：添加模糊测试支持
- apt-get和apt-cache是Ubuntu Linux中的命令行下的包管理工具。 apt-get的GUI版本是Synaptic包管理器。该文中我们会展示apt-get和apt-cache命令的15个不同例子。示例：1 列出所有可用包复制代码代码如下:linuxtechi@localhost:~$ apt-cache pkgnames account-plugin-yahoojp ceph-fuse dvd+rw-tools e3 gnome-commander-data grub-gfxpayload-lists gweled .......................................示例：2 用关键字搜索包这个命令在你不确定包名时很有用，只要在apt-cache（LCTT 译注：这里原文是apt-get，应为笔误）后面输入与包相关的关键字即可。复制代码代码如下:linuxtechi@localhost:~$ apt-cache search web server apache2 - Apache HTTP Server apache2-bin - Apache HTTP Server (binary files and modules) apache2-data - Apache HTTP Server (common files) apache2-dbg - Apache debugging symbols apache2-dev - Apache HTTP Server (development headers) apache2-doc - Apache HTTP Server (on-site documentation) apache2-utils - Apache HTTP Server (utility programs for web servers) ......................................................................注意： 假如你安装了“apt-file”包，我们就可以像下面那样用配置文件搜索包。复制代码代码如下:linuxtechi@localhost:~$ apt-file search nagios.cfg ganglia-nagios-bridge: /usr/share/doc/ganglia-nagios-bridge/nagios.cfg nagios3-common: /etc/nagios3/nagios.cfg nagios3-common: /usr/share/doc/nagios3-common/examples/nagios.cfg.gz pnp4nagios-bin: /etc/pnp4nagios/nagios.cfg pnp4nagios-bin: /usr/share/doc/pnp4nagios/examples/nagios.cfg示例：3 显示特定包的基本信息复制代码代码如下:linuxtechi@localhost:~$ apt-cache show postfix Package: postfix Priority: optional Section: mail Installed-Size: 3524 Maintainer: LaMont Jones Architecture: amd64 Version: 2.11.1-1 Replaces: mail-transport-agent Provides: default-mta, mail-transport-agent .....................................................示例：4 列出包的依赖复制代码代码如下:linuxtechi@localhost:~$ apt-cache depends postfix postfix Depends: libc6 Depends: libdb5.3 Depends: libsasl2-2 Depends: libsqlite3-0 Depends: libssl1.0.0 |Depends: debconf Depends: cdebconf debconf Depends: netbase Depends: adduser Depends: dpkg ............................................示例：5 使用apt-cache显示缓存统计复制代码代码如下:linuxtechi@localhost:~$ apt-cache stats Total package names: 60877 (1,218 k) Total package structures: 102824 (5,758 k) Normal packages: 71285 Pure virtual packages: 1102 Single virtual packages: 9151 Mixed virtual packages: 1827 Missing: 19459 Total distinct versions: 74913 (5,394 k) Total distinct descriptions: 93792 (2,251 k) Total dependencies: 573443 (16.1 M) Total ver/file relations: 78007 (1,872 k) Total Desc/File relations: 93792 (2,251 k) Total Provides mappings: 16583 (332 k) Total globbed strings: 171 (2,263 ) Total dependency version space: 2,665 k Total slack space: 37.3 k Total space accounted for: 29.5 M示例：6 使用 “apt-get update” 更新仓库使用命令“apt-get update”, 我们可以重新从源仓库中同步文件索引。包的索引从“/etc/apt/sources.list”中检索。复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get update Ign http://extras.ubuntu.com utopic InRelease Hit http://extras.ubuntu.com utopic Release.gpg Hit http://extras.ubuntu.com utopic Release Hit http://extras.ubuntu.com utopic/main Sources Hit http://extras.ubuntu.com utopic/main amd64 Packages Hit http://extras.ubuntu.com utopic/main i386 Packages Ign http://in.archive.ubuntu.com utopic InRelease Ign http://in.archive.ubuntu.com utopic-updates InRelease Ign http://in.archive.ubuntu.com utopic-backports InRelease ................................................................示例：7 使用apt-get安装包复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get install icinga上面的命令会安装叫“icinga”的包。示例：8 升级所有已安装的包复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get upgrade示例：9 更新特定的包在apt-get命令中的“install”选项后面接上“-only-upgrade”用来更新一个特定的包，如下所示：复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get install filezilla --only-upgrade示例：10 使用apt-get卸载包复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get remove skype上面的命令只会删除skype包，假如你想要删除它的配置文件，在apt-get命令中使用“purge”选项。如下所示：复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get purge skype我们可以结合使用上面的两个命令：复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get remove --purge skype示例：11 在当前的目录中下载包复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get download icinga Get:1 http://in.archive.ubuntu.com/ubuntu/ utopic/universe icinga amd64 1.11.6-1build1 [1,474 B] Fetched 1,474 B in 1s (1,363 B/s)上面的目录会把icinga包下载到你的当前工作目录。示例：12 清理本地包占用的磁盘空间复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get clean上面的命令会清空apt-get所下载的包占用的磁盘空间。我们也可以使用“autoclean”选项来代替“clean”，两者之间主要的区别是autoclean清理不再使用且没用的下载。复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get autoclean Reading package lists... Done Building dependency tree Reading state information... Done示例：13 使用“autoremove”删除包当在apt-get命令中使用“autoremove”时，它会删除为了满足依赖而安装且现在没用的包。复制代码代码如下: linuxtechi@localhost:~$ sudo apt-get autoremove icinga示例：14 显示包的更新日志复制代码代码如下: linuxtechi@localhost:~$ sudo apt-get changelog apache2 Get:1 Changelog for apache2 (http://changelogs.ubuntu.com/changelogs/pool/main/a/apache2/apache2_2.4.10-1ubuntu1/changelog) [195 kB] Fetched 195 kB in 3s (60.9 kB/s)上面的命令会下载apache2的更新日志，并在你屏幕上分页显示。示例：15 使用 “check” 选项显示损坏的依赖关系复制代码代码如下:linuxtechi@localhost:~$ sudo apt-get check Reading package lists... Done Building dependency tree Reading state information... Done
- 序列赋值引发的Python列表陷进
- 图解霍夫曼编码，教不会我吃一包辣条
- 代码调试是程序员必备技巧，简单示例，VS如何调试C++程序
- 华硕笔记本电脑安装系统教程（轻松学会如何在华硕笔记本电脑上安装系统）
- Java文件的简单读写、随机读写、NIO读写与使用MappedByteBuffer读写
随便看看
- Copyright © 2025 Powered by 如何在优雅地Spring中实现消息的发送和消费,码上建站 滇ICP备2023006006号-47sitemap