rocketmq-spring : 实战与源码解析一网打尽

本文涉及的产品
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
云解析 DNS,旗舰版 1个月
全局流量管理 GTM,标准版 1个月
简介: RocketMQ 是大家耳熟能详的消息队列,开源项目 rocketmq-spring 可以帮助开发者在 Spring Boot 项目中快速整合 RocketMQ。这篇文章会介绍 Spring Boot 项目使用 rocketmq-spring SDK 实现消息收发的操作流程,同时笔者会从开发者的角度解读 SDK 的设计逻辑。

RocketMQ 是大家耳熟能详的消息队列,开源项目 rocketmq-spring 可以帮助开发者在 Spring Boot 项目中快速整合 RocketMQ。

这篇文章会介绍 Spring Boot 项目使用 rocketmq-spring SDK 实现消息收发的操作流程,同时笔者会从开发者的角度解读 SDK 的设计逻辑

1 SDK 简介


项目地址:

https://github.com/apache/rocketmq-spring

rocketmq-spring 的本质是一个 Spring Boot starter

Spring Boot 基于“约定大于配置”(Convention over configuration)这一理念来快速地开发、测试、运行和部署 Spring 应用,并能通过简单地与各种启动器(如 spring-boot-web-starter)结合,让应用直接以命令行的方式运行,不需再部署到独立容器中。

Spring Boot starter 构造的启动器使用起来非常方便,开发者只需要在 pom.xml 引入 starter 的依赖定义,在配置文件中编写约定的配置即可。

下面我们看下 rocketmq-spring-boot-starter 的配置:

1、引入依赖

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-spring-boot-starter</artifactId>
  <version>2.2.3</version>
</dependency>
AI 代码解读

2、约定配置

接下来,我们分别按照生产者和消费者的顺序,详细的讲解消息收发的操作过程。

2 生产者

首先我们添加依赖后,进行如下三个步骤:

1、配置文件中配置如下

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
      group: platform-sms-server-group
    # access-key: myaccesskey
    # secret-key: mysecretkey
  topic: sms-common-topic
AI 代码解读

生产者配置非常简单,主要配置名字服务地址生产者组

2、需要发送消息的类中注入 RcoketMQTemplate

@Autowired
private RocketMQTemplate rocketMQTemplate;

@Value("${rocketmq.topic}")
private String smsTopic;
AI 代码解读

3、发送消息,消息体可以是自定义对象,也可以是 Message 对象

rocketMQTemplate 类包含多钟发送消息的方法:

  1. 同步发送 syncSend
  2. 异步发送 asyncSend
  3. 顺序发送 syncSendOrderly
  4. oneway发送 sendOneWay

下面的代码展示如何同步发送消息。

String destination = StringUtils.isBlank(tags) ? topic : topic + ":" + tags;
SendResult sendResult =
         rocketMQTemplate.syncSend(
            destination, 
            MessageBuilder.withPayload(messageContent).
            setHeader(MessageConst.PROPERTY_KEYS, uniqueId).
            build()
          );
if (sendResult != null) {
    if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
       // send message success ,do something 
    }
}
AI 代码解读

syncSend 方法的第一个参数是发送的目标,格式是:topic + ":" + tags

第二个参数是:spring-message 规范的 message 对象 ,而 MessageBuilder 是一个工具类,方法链式调用创建消息对象。

3 消费者

1、配置文件中配置如下

rocketmq:
  name-server: 127.0.0.1:9876
  consumer1:
    group: platform-sms-worker-common-group
    topic: sms-common-topic
AI 代码解读

2、实现消息监听器

@Component
@RocketMQMessageListener(
    consumerGroup = "${rocketmq.consumer1.group}",  //消费组
    topic = "${rocketmq.consumer1.topic}"                      //主题
)
public class SmsMessageCommonConsumer implements RocketMQListener<String> {
    public void onMessage(String message) {
       System.out.println("普通短信:" + message);
    }
}
AI 代码解读

消费者实现类也可以实现 RocketMQListener\<MessageExt>, 在 onMessage 方法里通过 RocketMQ 原生消息对象 MessageExt 获取更详细的消息数据

public void onMessage(MessageExt message) {
    try {
        String body = new String(message.getBody(), "UTF-8");
        logger.info("普通短信:" + message);
    } catch (Exception e) {
        logger.error("common onMessage error:", e);
    }
}
AI 代码解读

4 源码概览

最新源码中,我们可以看到源码中包含四个模块:

1、rocketmq-spring-boot-parent

该模块是父模块,定义项目所有依赖的 jar 包。

2、rocketmq-spring-boot

核心模块,实现了 starter 的核心逻辑。

3、rocketmq-spring-boot-starter

SDK 模块,简单封装,外部项目引用。

4、rocketmq-spring-boot-samples

示例代码模块。这个模块非常重要,当用户使用 SDK 时,可以参考示例快速开发。

5 starter 实现

我们重点分析下 rocketmq-spring-boot 模块的核心源码:

<br/>

spring-boot-starter 实现需要包含如下三个部分:

1、定义 Spring 自身的依赖包和 RocketMQ 的依赖包 ;

2、定义spring.factories 文件

在 resources 包下创建 META-INF 目录后,新建 spring.factories 文件,并在文件中定义自动加载类,文件内容是:

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration
AI 代码解读

spring boot 会根据文件中配置的自动化配置类来自动初始化相关的 Bean、Component 或 Service。

3、实现自动加载类

在 RocketMQAutoConfiguration 类的具体实现中,我们重点分析下生产者和消费者是如何分别启动的。

▍生产者发送模板类:RocketMQTemplate

RocketMQAutoConfiguration 类定义了两个默认的 Bean :

首先SpringBoot项目中配置文件中的配置值会根据属性条件绑定到 RocketMQProperties 对象 中,然后使用 RocketMQ 的原生 API 分别创建生产者 Bean 和拉取消费者 Bean , 分别将两个 bean 设置到 RocketMQTemplate 对象中。

两个重点需要强调:

  • 发送消息时,将 spring-message 规范下的消息对象封装成 RocketMQ 消息对象

  • 默认拉取消费者 litePullConsumer 。拉取消费者一般用于大数据批量处理场景 。

    原生使用方式

​ RocketMQTemplate 类封装了拉取消费者的receive方法,以方便开发者使用。

▍自定义消费者类

下图是并发消费者的例子:

消费者示例代码

那么 rocketmq-spring 是如何自动启动消费者呢 ?

spring 容器首先注册了消息监听器后置处理器,然后调用 ListenerContainerConfiguration 类的 registerContainer 方法 。

对比并发消费者的例子,我们可以看到: DefaultRocketMQListenerContainer 是对 DefaultMQPushConsumer 消费逻辑的封装。

封装消费消息的逻辑,同时满足 RocketMQListener 泛化接口支持不同参数,比如 String 、MessageExt 、自定义对象 。

首先DefaultRocketMQListenerContainer初始化之后, 获取 onMessage 方法的参数类型 。

然后消费者调用 consumeMessage 处理消息时,封装了一个 handleMessage 方法 ,将原生 RocketMQ 消息对象 MessageExt 转换成 onMessage 方法定义的参数对象,然后调用 rocketMQListener 的 onMessage 方法。

mnjh9

上图右侧标红的代码也就是该方法的精髓:

rocketMQListener.onMessage(doConvertMessage(messageExt));
AI 代码解读

6 写到最后

开源项目 rocketmq-spring 有很多值得学习的地方 ,我们可以从如下四个层面逐层进阶:

1、学会如何使用 :参考 rocketmq-spring-boot-samples 模块的示例代码,学会如何发送和接收消息,快速编码;

2、模块设计:学习项目的模块分层 (父模块、SDK 模块、核心实现模块、示例代码模块);

3、starter 设计思路 :定义自动配置文件 spring.factories 、设计配置属性类 、在 RocketMQ client 的基础上实现优雅的封装、深入理解 RocketMQ 源码等;

4、举一反三:当我们理解了 rocketmq-spring 的源码,我们可以尝试模仿该项目写一个简单的 spring boot starter。


如果我的文章对你有所帮助,还请帮忙点赞、在看、转发一下,你的支持会激励我输出更高质量的文章,非常感谢!

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
打赏
0
0
0
0
251
分享
相关文章
Jeesite5:Star24k,Spring Boot 3.3+Vue3实战开源项目,架构深度拆解!让企业级项目开发效率提升300%的秘密武器
Jeesite5 是一个基于 Spring Boot 3.3 和 Vue3 的企业级快速开发平台,集成了众多优秀开源项目,如 MyBatis Plus、Bootstrap、JQuery 等。它提供了模块化设计、权限管理、多数据库支持、代码生成器和国际化等功能,极大地提高了企业级项目的开发效率。Jeesite5 广泛应用于企业管理系统、电商平台、客户关系管理和知识管理等领域。通过其强大的功能和灵活性,Jeesite5 成为了企业级开发的首选框架之一。访问 [Gitee 页面](https://gitee.com/thinkgem/jeesite5) 获取更多信息。
Jeesite5:Star24k,Spring Boot 3.3+Vue3实战开源项目,架构深度拆解!让企业级项目开发效率提升300%的秘密武器
|
27天前
|
Spring底层架构核心概念解析
理解 Spring 框架的核心概念对于开发和维护 Spring 应用程序至关重要。IOC 和 AOP 是其两个关键特性,通过依赖注入和面向切面编程实现了高效的模块化和松耦合设计。Spring 容器管理着 Beans 的生命周期和配置,而核心模块为各种应用场景提供了丰富的功能支持。通过全面掌握这些核心概念,开发者可以更加高效地利用 Spring 框架开发企业级应用。
80 18
Python执行Shell命令并获取结果:深入解析与实战
通过以上内容,开发者可以在实际项目中灵活应用Python执行Shell命令,实现各种自动化任务,提高开发和运维效率。
61 20
深度解析1688 API对电商的影响与实战应用
在全球电子商务迅猛发展的背景下,1688作为知名的B2B电商平台,为中小企业提供商品批发、分销、供应链管理等一站式服务,并通过开放的API接口,为开发者和电商企业提供数据资源和功能支持。本文将深入解析1688 API的功能(如商品搜索、详情、订单管理等)、应用场景(如商品展示、搜索优化、交易管理和用户行为分析)、收益分析(如流量增长、销售提升、库存优化和成本降低)及实际案例,帮助电商从业者提升运营效率和商业收益。
208 20
智慧工地云平台的技术架构解析:微服务+Spring Cloud如何支撑海量数据?
慧工地解决方案依托AI、物联网和BIM技术,实现对施工现场的全方位、立体化管理。通过规范施工、减少安全隐患、节省人力、降低运营成本,提升工地管理的安全性、效率和精益度。该方案适用于大型建筑、基础设施、房地产开发等场景,具备微服务架构、大数据与AI分析、物联网设备联网、多端协同等创新点,推动建筑行业向数字化、智能化转型。未来将融合5G、区块链等技术,助力智慧城市建设。
【23种设计模式·全精解析 | 自定义Spring框架篇】Spring核心源码分析+自定义Spring的IOC功能,依赖注入功能
本文详细介绍了Spring框架的核心功能,并通过手写自定义Spring框架的方式,深入理解了Spring的IOC(控制反转)和DI(依赖注入)功能,并且学会实际运用设计模式到真实开发中。
【23种设计模式·全精解析 | 自定义Spring框架篇】Spring核心源码分析+自定义Spring的IOC功能,依赖注入功能
【23种设计模式·全精解析 | 行为型模式篇】11种行为型模式的结构概述、案例实现、优缺点、扩展对比、使用场景、源码解析
行为型模式用于描述程序在运行时复杂的流程控制,即描述多个类或对象之间怎样相互协作共同完成单个对象都无法单独完成的任务,它涉及算法与对象间职责的分配。行为型模式分为类行为模式和对象行为模式,前者采用继承机制来在类间分派行为,后者采用组合或聚合在对象间分配行为。由于组合关系或聚合关系比继承关系耦合度低,满足“合成复用原则”,所以对象行为模式比类行为模式具有更大的灵活性。 行为型模式分为: • 模板方法模式 • 策略模式 • 命令模式 • 职责链模式 • 状态模式 • 观察者模式 • 中介者模式 • 迭代器模式 • 访问者模式 • 备忘录模式 • 解释器模式
【23种设计模式·全精解析 | 行为型模式篇】11种行为型模式的结构概述、案例实现、优缺点、扩展对比、使用场景、源码解析
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析
结构型模式描述如何将类或对象按某种布局组成更大的结构。它分为类结构型模式和对象结构型模式,前者采用继承机制来组织接口和类,后者釆用组合或聚合来组合对象。由于组合关系或聚合关系比继承关系耦合度低,满足“合成复用原则”,所以对象结构型模式比类结构型模式具有更大的灵活性。 结构型模式分为以下 7 种: • 代理模式 • 适配器模式 • 装饰者模式 • 桥接模式 • 外观模式 • 组合模式 • 享元模式
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析
深入解析BeautifulSoup:从sohu.com视频页面提取关键信息的实战技巧
深入解析BeautifulSoup:从sohu.com视频页面提取关键信息的实战技巧
速卖通AliExpress商品详情API接口深度解析与实战应用
速卖通(AliExpress)作为全球化电商的重要平台,提供了丰富的商品资源和便捷的购物体验。为了提升用户体验和优化商品管理,速卖通开放了API接口,其中商品详情API尤为关键。本文介绍如何获取API密钥、调用商品详情API接口,并处理API响应数据,帮助开发者和商家高效利用这些工具。通过合理规划API调用策略和确保合法合规使用,开发者可以更好地获取商品信息,优化管理和营销策略。

推荐镜像

更多
AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等