当前位置: 首页 > article >正文

20241130 RocketMQ本机安装与SpringBoot整合

目录

一、RocketMQ简介

???1.1、核心概念

???1.2、应用场景

???1.3、架构设计

2、RocketMQ Server安装

3、RocketMQ可视化控制台安装与使用

4、SpringBoot整合RocketMQ实现消息发送和接收?

? ? ? ? 4.1、添加maven依赖

???4.2、yaml配置

???4.3、生产者

???4.4、消费者

???4.5、接口

???4.6、接口测试


一、RocketMQ****简介

RocketMQ是由阿里捐赠给Apache的一款低延迟、高并发、高可用、高可靠的分布式消息中间件。经历 了淘宝双十一的洗礼。RocketMQ既可为分布式应用系统提供异步解耦和削峰填谷的能力,同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。

官方文档:https://rocketmq.apache.org/docs/quick-start/

github中文主页:https://github.com/apache/rocketmq/tree/master/docs/cn

1.1、核心概念

  1. Topic:消息主题,一级消息类型,生产者向其发送消息。Message:生产者向Topic发送并最终传送给消费者的数据消息的载体。
  2. 消息属性:生产者可以为消息定义的属性,包含Message Key和Tag。
  3. Message Key:消息的业务标识,由消息生产者(Producer)设置,唯一标识某个业务逻辑。
  4. Message ID:消息的全局唯一标识,由消息队列RocketMQ系统自动生成,唯一标识某条消息。
  5. Tag:消息标签,二级消息类型,用来进一步区分某个Topic下的消息分类
  6. Producer:也称为消息发布者,负责生产并发送消息至Topic。
  7. Consumer:也称为消息订阅者,负责从Topic接收并消费消息。
  8. 分区:即Topic Partition,物理上的概念。每个Topic包含一个或多个分区。
  9. 消费位点:每个Topic会有多个分区,每个分区会统计当前消息的总条数,这个称为最大位点MaxOffset;分区的起始位置对应的位置叫做起始位点MinOffset。
  10. Group:一类生产者或消费者,这类生产者或消费者通常生产或消费同一类消息,且消息发布或订 阅的逻辑一致。
  11. Group ID:Group的标识。
  12. 队列:单个Topic下会由一到多个队列来存储消息。
  13. Exactly-Once****投递语义:Exactly-Once投递语义是指发送到消息系统的消息只能被Consumer处理且仅处理一次,即使Producer重试消息发送导致某消息重复投递,该消息在Consumer也只被消费 一次。
  14. 集群消费:一个Group ID所标识的所有Consumer平均分摊消费消息。例如某个Topic有9条消息,一个Group ID有3个Consumer实例,那么在集群消费模式下每个实例平均分摊,只消费其中的3条 消息。
  15. 广播消费:一个Group ID所标识的所有Consumer都会各自消费某条消息一次。例如某个Topic有9条消息,一个Group ID有3个Consumer实例,那么在广播消费模式下每个实例都会各自消费9条消息。
  16. 定时消息:Producer将消息发送到消息队列RocketMQ服务端,但并不期望这条消息立马投递,而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费,该消息即定时消息。
  17. 延时消息:Producer将消息发送到消息队列RocketMQ服务端,但并不期望这条消息立马投递,而是延迟一定时间后才投递到Consumer进行消费,该消息即延时消息。
  18. 事务消息:RocketMQ提供类似X/Open XA的分布事务功能,通过消息队列RocketMQ的事务消息能达到分布式事务的最终一致。
  19. 顺序消息:RocketMQ提供的一种按照顺序进行发布和消费的消息类型,分为全局顺序消息和分区顺序消息。
  20. 全局顺序消息:对于指定的一个Topic,所有消息按照严格的先入先出(FIFO)的顺序进行发布和消费。
  21. 分区顺序消息:对于指定的一个Topic,所有消息根据Sharding Key进行区块分区。同一个分区内的消息按照严格的FIFO顺序进行发布和消费。Sharding Key是顺序消息中用来区分不同分区的关键字段,和普通消息的Message Key是完全不同的概念。
  22. 消息堆积:Producer已经将消息发送到消息队列RocketMQ的服务端,但由于Consumer消费能力有限,未能在短时间内将所有消息正确消费掉,此时在消息队列RocketMQ的服务端保存着未被消费的消息,该状态即消息堆积。
  23. 消息过滤:Consumer可以根据消息标签(Tag)对消息进行过滤,确保Consumer最终只接收被过滤后的消息类型。消息过滤在消息队列RocketMQ的服务端完成。
  24. 消息轨迹:在一条消息从Producer发出到Consumer消费处理过程中,由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹,您能清晰定位消息从Producer发出,经由消息队列RocketMQ服务端,投递给Consumer的完整链路,方便定位排查问题。
  25. 重置消费位点:以时间轴为坐标,在消息持久化存储的时间范围内(默认3天),重新设置Consumer对已订阅的Topic的消费进度,设置完成后Consumer将接收设定时间点之后由Producer发送到消息队列RocketMQ服务端的消息。
  26. 死信队列:死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败,消息队列RocketMQ会自动进行消息重试;达到最大重试次数后,若消费依然失败,则表明Consumer在正常情况下无法正确地消费该消息。此时,消息队列RocketMQ不会立刻将消息丢弃,而是将这条消息发送到该Consumer对应的特殊队列中。消息队RocketMQ将这种正常情况下无法被消费的消息称为死信消息(Dead-Letter Message),将存储死信消息的特殊队列称为死信队列(Dead-Letter Queue)。

1.2、应用场景

  1. 削峰填谷:诸如秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲,或因没做相应的保护而导致系统超负荷甚至崩溃,或因限制太过导致请求大量失败而影响用户体验,消息队列RocketMQ可提供削峰填谷的服务来解决该问题。
  2. 异步解耦:交易系统作为淘宝和天猫主站最核心的系统,每笔交易订单数据的产生会引起几百个下游业务系统的关注,包括物流、购物车、积分、流计算分析等等,整体业务系统庞大而且复杂,消息队列RocketMQ可实现异步通信和应用解耦,确保主站业务的连续性。
  3. 顺序收发:细数日常中需要保证顺序的应用场景非常多,例如证券交易过程时间优先原则,交易系统中的订单创建、支付、退款等流程,航班中的旅客登机消息处理等等。与先进先出FIFO(First In First Out)原理类似,消息队列RocketMQ提供的顺序消息即保证消息FIFO。
  4. 分布式事务一致性:交易系统、支付红包等场景需要确保数据的最终一致性,大量引入消息队列RocketMQ的分布式事务,既可以实现系统之间的解耦,又可以保证最终的数据一致性。
  5. 大数据分析:数据在“流动”中产生价值,传统数据分析大多是基于批量计算模型,而无法做到实时的数据分析,利用阿里云消息队列RocketMQ与流式计算引擎相结合,可以很方便的实现业务数据的实时分析。
  6. 分布式缓存同步:天猫双11大促,各个分会场琳琅满目的商品需要实时感知价格变化,大量并发访问数据库导致会场页面响应时间长,集中式缓存因带宽瓶颈,限制了商品变更的访问流量,通过消息队列RocketMQ构建分布式缓存,实时通知商品数据的变化。

1.3、架构设计

RocketMQ架构上主要分为四部分,如上图所示:

**Producer:**消息发布的角色,支持分布式集群方式部署。 Producer 通过 MQ 的负载均衡模块选择相应的Broker集群队列进行消息投递,投递的过程支持快速失败并且低延迟。

**Consumer:**消息消费的角色,支持分布式集群方式部署。支持以 push 推, pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费,它提供实时消息订阅机制,可以满足大多数用户的需求。

NameServer: NameServer 是一个非常简单的 Topic 路由注册中心,其角色类似 Dubbo 中的zookeeper,支持 Broker 的动态注册与发现。主要包括两个功能: Broker 管理, NameServer 接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制,检查Broker是否还存活;路由信息管理,每个 NameServer 将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。然后Producer 和 Conumser 通过 NameServer 就可以知道整个 Broker 集群的路由信息,从而进行消息的投递和消费。NameServer 通常也是集群的方式部署,各实例间相互不进行信息通讯。Broker 是向每一台 NameServer 注册自己的路由信息,所以每一个NameServer实例上面都保存一份完整的路由信息。当某个 NameServer 因某种原因下线了,Broker仍然可以向其它 NameServer 同步其路由信息, Producer,Consumer仍然可以动态感知

Broker的路由的信息。BrokerServer: Broker 主要负责消息的存储、投递和查询以及服务高可用保证,为了实现这些功能,Broker包含了以下几个重要子模块。

**1. Remoting Module:**整个Broker 的实体,负责处理来自 clients端的请求。

**2. Client Manager:**负责管理客户端(Producer/Consumer) 和维护 Consumer 的 Topic订阅信息

**3. Store Service:**提供方便简单的API接口处理消息存储到物理硬盘和查询功能。

4. HA Service: 高可用服务,提供Master Broker 和 Slave Broker之间的数据同步功能。

5. Index Service: 根据特定的Message key 对投递到 Broker的消息进行索引服务,以提供消息的快速查询。

2、RocketMQ Server****安装

RocketMQ依赖Java环境,要求有JDK 1.8以上版本;

支持Windows和Linux平台;支持源码方式安装和使用已经编译好的安装包安装;

我们用windows平台安装RocketMQ Server编译好的安装包,来讲解RocketMQ;

下载地址:https://rocketmq.apache.org/dowloading/releases/

我们选择4.9.0二进制版本;

解压后的目录:

benchmark:里面是测试Demo;

bin:可执行脚本;conf:配置文件;

lib:依赖的jar包;

我们把rocketmq解压包放到D盘soft目录下,重命名 rocketmq ;

第一步:系统环境变量加两个配置

ROCKETMQ_HOME=“D:soft ocketmq”

NAMESRV_ADDR=“localhost:9876”

第二步:启动 Name Server

进入命令行执行:

.inmqnamesrv.cmd

第三步:启动 Broker

进入命令行执行:

.inmqbroker.cmd -n localhost:9876 autoCreateTopicEnable=true

第四步:发送和接收消息测试

进入命令行消息发送执行:

.in ools.cmd org.apache.rocketmq.example.quickstart.Producer

消息发送成功;

进入命令行消息接收执行:

.in ools.cmd org.apache.rocketmq.example.quickstart.Consumer

消息接收成功;

第五步:关闭服务

windows 下直接关闭命令行窗口即可;

3、RocketMQ****可视化控制台安装与使用

RocketMQ 提供了一些扩展项目支持,地址: https://github.com/apache/rocketmq-externals

其中一个 rocketmq - connect - console 项目,就是我们需要的可视化控制台;

我们把整个项目下载下来,打开 rocketmq - console 项目;项目是 SpringBoot 开发;

打开 application.properties 配置文件,我们至少需要修改两个配置项;

server.port=8080 ,这个是可视化项目启动端口,我们改成 19876 ;

rocketmq.config.namesrvAddr= ,这个是指定 nameServer 地址和端口,我们暂时先搞成localhost:9876,等后面搞集群的话,要再修改;

修改后保存,在项目目录下进入命令行,执行:

D:soft ocketmqClient ocketmq-externals-master ocketmq-console

mvn clean package -Dmaven.test.skip=true

打包执行完后,在 target 目录,会生成一个可运行 jar rocketmq - console - ng - 2.0.0.jar

我们运行这个 jar ,进入命令行执行:

java -jar rocketmq-console-ng-2.0.0.jar

启动成功后,浏览器输入:http://localhost:19876/

说明一切OK;

4、SpringBoot整合RocketMQ****实现消息发送和接收

4.1、添加maven依赖

org.apache.rocketmq rocketmq-spring-boot-starter 2.2.3

4.2、yaml配置

rocketmq:
name-server: localhost:9876
producer:
# 发送同一类消息的设置为同一个group,保证唯一
group: producer-group
# 发送消息超时时间,默认3000
sendMessageTimeout: 5000
# 发送消息失败重试次数,默认2
retryTimesWhenSendFailed: 2
# 异步消息重试此处,默认2
retryTimesWhenSendAsyncFailed: 2
# 消息最大长度,默认1024 * 1024 * 4(默认4M)
maxMessageSize: 4096
# 压缩消息阈值,默认4k(1024 * 4)
compressMessageBodyThreshold: 4096
# 是否在内部发送失败时重试另一个broker,默认false
retryNextServer: false
consumer:
group: consumer-group # 消费者组名

4.3、生产者

@Service
public class RocketMQProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, String message) {
        rocketMQTemplate.convertAndSend(topic, message);
        System.out.println("Message sent: " + message);
    }
}

4.4、消费者

@Service
@RocketMQMessageListener(topic = "your-topic", consumerGroup = "consumer-group")
public class RocketMQConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

4.5、接口

    @GetMapping("/send")
    public String sendMessage(@RequestParam String topic, @RequestParam String message) {
        rocketMQProducer.sendMessage(topic, message);
        return "Message sent: " + message;
    }

4.6、接口测试

至此测试完成!


http://www.kler.cn/a/506691.html

相关文章:

  • HarmonyOS Next 实现登录注册页面(ARKTS) 并使用Springboot作为后端提供接口
  • Git 合并和 Git 变基有什么区别?
  • 基于华为云车牌识别服务设计的停车场计费系统【华为开发者空间-鸿蒙】
  • C语言初阶习题【30】字符串左旋
  • STC的51单片机LED点灯基于KEIL
  • VSCode连接Github的重重困难及解决方案!
  • CVPR 2024 视频处理方向总汇(视频监控、视频理解、视频识别和视频预测等)
  • JavaScript系列(29)--设计模式详解
  • 2025年01月15日Github流行趋势
  • 火绒剑独立版 - 强大的Windows系统安全分析工具
  • 基于 Python 的毕设选题管理系统设计与实现
  • 前端如何创建微任务
  • 【gRPC】clientPool 客户端连接池简单实现与go案例
  • Go语言之路————条件控制:if、for、switch
  • Oracle EBS GL定期盘存WIP日记账无法过账数据修复
  • Go语言封装加解密包(AES/DES/RSA)
  • Sprint Boot教程之五十八:动态启动/停止 Kafka 监听器
  • 说说Babylon.js中scene.deltaTime的大坑
  • 如何异地远程访问本地部署的Web-Check实现团队远程检测与维护本地站点
  • 《DeepSeek V3:重新定义AI大模型的效率与成本》
  • Qt实现防止程序多次运行
  • Java学习教程,从入门到精通,JDBC数据库连接语法知识点及案例代码(92)
  • Outlook 无网络连接[2604] 错误解决办法
  • python批量doc转pdf调用提示库未注册
  • 华北水利水电大学第十届ACM/ICPC程序设计新生赛题解
  • Django Admin 实战:实现 ECS 集群批量同步功能