微服务轮子项目(33) -RocketMQ特点、安装部署、异常处理、事务消息原理

简介: 微服务轮子项目(33) -RocketMQ特点、安装部署、异常处理、事务消息原理

1. Rocket概述

消息队列作为高并发系统的核心组件之一,能够帮助业务系统解构提升开发效率和系统稳定性。主要具有以下优势:

  • 削峰填谷:主要解决瞬时写压力大于应用服务能力导致消息丢失、系统奔溃等问题
  • 系统解耦:解决不同重要程度、不同能力级别系统之间依赖导致一死全死
  • 提升性能:当存在一对多调用时,可以发一条消息给消息系统,让消息系统通知相关系统
  • 蓄流压测:线上有些链路不好压测,可以通过堆积一定量消息再放开来压测

1.1 特点

Apache Alibaba RocketMQ 是一款分布式、队列模型的消息中间件,具有以下特点:

  • 支持严格的消息顺序
  • 支持 Topic 与 Queue 两种模式
  • 亿级消息堆积能力
  • 比较友好的分布式特性
  • 同时支持 Push 与 Pull 方式消费消息
  • 历经多次天猫双十一海量消息考验

目前主流的 MQ 主要是 RocketMQ、kafka、RabbitMQ,对比其主要优势有:

  • 支持事务型消息(消息发送和 DB 操作保持两方的最终一致性,RabbitMQ 和 Kafka 不支持)
  • 支持结合 RocketMQ 的多个系统之间数据最终一致性(多方事务,二方事务是前提)
  • 支持 18 个级别的延迟消息(RabbitMQ 和 Kafka 不支持)
  • 支持指定次数和时间间隔的失败消息重发(Kafka 不支持,RabbitMQ 需要手动确认)
  • 支持 Consumer 端 Tag 过滤,减少不必要的网络传输(RabbitMQ 和 Kafka 不支持)
  • 支持重复消费(RabbitMQ 不支持,Kafka 支持)

1.2 网络部署架构

  • NameServer:是一个几乎无状态的节点,可集群部署,节点之间无任何信息同步
  • Broker:部署相对复杂,Broker分为Master与Slave,一个Master可以对应多个Slaver,但是一个Slaver只能对应一个Master,Master与Slaver的对应关系通过指定相同的BrokerName,不同的BrokerId来定义,BrokerId为0表示Master,非0表示Slaver。Master可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接,定时注册Topic信息到所有的NameServer
  • Producer:与NameServer集群中的其中一个节点(随机选择)建立长连接,定期从NameServer取Topic路由信息,并向提供Topic服务的Master建立长连接,且定时向Master发送心跳。Produce完全无状态,可集群部署
  • Consumer:与NameServer集群中的其中一个节点(随机选择)建立长连接,定期从NameServer取Topic路由信息,并向提供Topic服务的Master、Slaver建立长连接,且定时向Master、Slaver发送心跳。Consumer即可从Master订阅消息,也可以从Slave订阅消息,订阅规则由Broker配置决定

1.3 存储特点

1.3.1 零拷贝原理

Consumer消费消息过程,使用了零拷贝,零拷贝包括以下两种方式,RocketMQ使用第一种方式,因小块数据传输的要求效果比sendfile方式好。

使用mmap+write方式

  • 优点:即使频繁调用,使用小文件块传输,效率也很高
  • 缺点:不能很好的利用DMA方式,会比sendfile多消耗CPU资源,内存安全性控制复杂,需要避免JVM Crash问题

使用sendfile方式

  • 优点:可以利用DMA方式,消耗CPU资源少,大块文件传输效率高,无内存安全新问题
  • 缺点:小块文件效率低于mmap方式,只能是BIO方式传输,不能使用NIO
1.3.2 数据存储结构

1.3.3 顺序消息和幂等性

顺序消息原理producer在发送消息的时候,把消息发到同一个队列(queue)中,消费者注册消息监听器为MessageListenerOrderly,这样就可以保证消费端只有一个线程去消费消息。(注意:把消息发到同一个队列(queue),不是同一个topic,默认情况下一个topic包括4个queue

顺序消息缺陷:发送顺序消息无法利用集群Fail Over特性消费顺序消息的并行度依赖于队列数量队列热点问题,个别队列由于哈希不均导致消息过多,消费速度跟不上,产生消息堆积问题遇到消息失败的消息,无法跳过,当前队列消费暂停。

消息幂等性RocketMQ使用的消息原语是At Least Once,所以consumer可能多次收到同一个消息,如果业务需要保证严格的不重复消息时需要自己做好幂等。

造成消息重复的根本原因是:网络不可达。只要通过网络交换数据,就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了:如果消费端收到两条一样的消息,应该怎样处理?

  • 消费端处理消息的业务逻辑保持幂等性
  • 保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现
1.3.4 相关资料

2. 安装部署

2.1 下载

下载地址:http://rocketmq.apache.org/release_notes/release-notes-4.5.0/

可以直接下载编译好的

2.2 修改配置

1.修改conf/broker.conf,添加以下配置:

brokerIP1:配置broker所在服务器的ip地址,以便Name Server连接

2.修改runserver.shrunbroker.sh(可不改),因为rocketMQ默认的启动参数内存占用非常大,如果环境没有这么多内存就必需修改JAVA_OPT参数:

runserver.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn125m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"


runbroker.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn125m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m

2.3 运行

2.3.1 运行Name Server
nohup sh bin/mqnamesrv &

查看运行日志:tail -f ~/logs/rocketmqlogs/broker.log

2.3.2 运行Broker
nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &
• 1

通过-c参数指定配置文件

查看运行日志:tail -f ~/logs/rocketmqlogs/broker.log

2.4 安装可视化管理界面

下载:

https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console

编译:

mvn clean package -Dmaven.test.skip=true

运行:

nohup java -jar \
    -Drocketmq.config.namesrvAddr=192.168.28.130:9876 \
    -Drocketmq.config.isVIPChannel=false \
    rocketmq-console-ng-1.0.0.jar &

访问:

http://192.168.28.130:8080/

2.5 停止服务

如果需要停止rocketMQ的服务,在生产环境不建议直接用kill,应该使用以下命令:

sh bin/mqshutdown broker
sh bin/mqshutdown namesrv

3. 常见异常处理

3.1 MQClientException: No route info of this topic, TopicTest1

在客户端的 Producer 运行起来准备发送消息时抛异常为 “ No route info of this topic” 异常产生的原因可能是:

  • Broker 禁止自动创建 Topic,且用户没有通过手工方式创建 Topic
  • Broker没有正确连接到Name Server
  • Producer没有正确连接到Name Server

解决办法:

排查1:Broker 禁止自动创建 Topic,且用户没有通过手工方式创建 Topic

  • 可以在rocketmq所在目录下执行" sh bin/mqbroker -m " 来查看 broker的配置参数
  • 如下所示,autoCreateTopicEnable=true 证明是没有问题的

排查2:Broker 没有正确连接到 Name Server

  • 通过查看broker的日志tail -f ~/logs/rocketmqlogs/broker.log看看有没有错误信息

排查3:Producer 没有正确连接到 Name Server

  • 检查程序连接Name Server的地址有没有错
  • 如果在云服务器上,检查安全组的配置9876端口有没有开发
  • 看看有没有打开防火墙,有的话设置防火墙开放9876端口
[root@zlt rocketmq-all-4.5.0-bin-release]# firewall-cmd --zone=public --list-ports
8090/tcp 80/tcp 8080/tcp
[root@zlt rocketmq-all-4.5.0-bin-release]# firewall-cmd --zone=public --add-port=9876/tcp --permanent
success
[root@zlt rocketmq-all-4.5.0-bin-release]# firewall-cmd --reload
success
[root@zlt rocketmq-all-4.5.0-bin-release]# firewall-cmd --zone=public --list-ports
9876/tcp 8090/tcp 80/tcp 8080/tcp
• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8

3.2 RemotingTooMuchRequestException: sendDefaultImpl call timeout

在客户端的 Producer 运行起来准备发送消息时抛异常如下,通常因为Name Server连接不上Broker

解决办法:

  • 检查rocketmq-console的集群页签,broker的地址是否正确:

broker地址的配置方式请参考安装部署中提到的步骤:

  • 修改broker.conf的配置,添加brokerIP1参数
  • 启动broker时加上-c参数指定配置文件

3.3 消费/查看不了死信队列topic的消息

死信队列默认的perm值为2没有查看权限

解决办法:

  • 在控制台把队列的perm改为6就可以了

4. RocketMQ事务消息原理

4.1 原理图

分为两个逻辑:正常事务消息的发送及提交、事务消息的回查流程

事务消息发送及提交

  • 发送消息(half消息)
  • 服务端响应消息写入结果
  • 根据发送结果执行本地事务(如果写入失败,此时half消息对业务不可见,本地逻辑不执行)
  • 根据本地事务状态执行Commit或者Rollback(Commit操作生成消息索引,消息对消费者可见)

回查流程

  • 对没有Commit/Rollback的事务消息(pending状态的消息),从服务端发起一次“回查”
  • Producer收到回查消息,检查回查消息对应的本地事务的状态
  • 根据本地事务状态,重新Commit或者Rollback

4.2 时序图


相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
Java Maven Android开发
微服务——SpringBoot使用归纳——Spring Boot开发环境搭建和项目启动
本文介绍了Spring Boot开发环境的搭建和项目启动流程。主要内容包括:jdk的配置(IDEA、STS/eclipse设置方法)、Spring Boot工程的构建方式(IDEA快速构建、官方构建工具start.spring.io使用)、maven配置(本地maven路径与阿里云镜像设置)以及编码配置(IDEA和eclipse中的编码设置)。通过这些步骤,帮助开发者顺利完成Spring Boot项目的初始化和运行准备。
1399 0
微服务——SpringBoot使用归纳——Spring Boot开发环境搭建和项目启动
|
Java 关系型数据库 数据库
微服务——SpringBoot使用归纳——Spring Boot事务配置管理——常见问题总结
本文总结了Spring Boot中使用事务的常见问题,虽然通过`@Transactional`注解可以轻松实现事务管理,但在实际项目中仍有许多潜在坑点。文章详细分析了三个典型问题:1) 异常未被捕获导致事务未回滚,需明确指定`rollbackFor`属性;2) 异常被try-catch“吃掉”,应避免在事务方法中直接处理异常;3) 事务范围与锁范围不一致引发并发问题,建议调整锁策略以覆盖事务范围。这些问题看似简单,但一旦发生,排查难度较大,因此开发时需格外留意。最后,文章提供了课程源代码下载地址,供读者实践参考。
512 0
|
Java 关系型数据库 数据库
微服务——SpringBoot使用归纳——Spring Boot事务配置管理——Spring Boot 事务配置
本文介绍了 Spring Boot 中的事务配置与使用方法。首先需要导入 MySQL 依赖,Spring Boot 会自动注入 `DataSourceTransactionManager`,无需额外配置即可通过 `@Transactional` 注解实现事务管理。接着通过创建一个用户插入功能的示例,展示了如何在 Service 层手动抛出异常以测试事务回滚机制。测试结果表明,数据库中未新增记录,证明事务已成功回滚。此过程简单高效,适合日常开发需求。
1980 0
|
Java 数据库 微服务
微服务——SpringBoot使用归纳——Spring Boot事务配置管理——事务相关
本文介绍Spring Boot事务配置管理,阐述事务在企业应用开发中的重要性。事务确保数据操作可靠,任一异常均可回滚至初始状态,如转账、购票等场景需全流程执行成功才算完成。同时,事务管理在Spring Boot的service层广泛应用,但根据实际需求也可能存在无需事务的情况,例如独立数据插入操作。
400 0
|
Java 测试技术 微服务
微服务——SpringBoot使用归纳——Spring Boot中的项目属性配置——少量配置信息的情形
本课主要讲解Spring Boot项目中的属性配置方法。在实际开发中,测试与生产环境的配置往往不同,因此不应将配置信息硬编码在代码中,而应使用配置文件管理,如`application.yml`。例如,在微服务架构下,可通过配置文件设置调用其他服务的地址(如订单服务端口8002),并利用`@Value`注解在代码中读取这些配置值。这种方式使项目更灵活,便于后续修改和维护。
466 0
|
Java 微服务 Spring
微服务——SpringBoot使用归纳——Spring Boot使用slf4j进行日志记录——使用Logger在项目中打印日志
本文介绍了如何在项目中使用Logger打印日志。通过SLF4J和Logback,可设置不同日志级别(如DEBUG、INFO、WARN、ERROR)并支持占位符输出动态信息。示例代码展示了日志在控制器中的应用,说明了日志配置对问题排查的重要性。附课程源码下载链接供实践参考。
1545 0
|
Java 数据库 微服务
微服务——SpringBoot使用归纳——Spring Boot中的项目属性配置——指定项目配置文件
在实际项目中,开发环境和生产环境的配置往往不同。为简化配置切换,可通过创建 `application-dev.yml` 和 `application-pro.yml` 分别管理开发与生产环境配置,如设置不同端口(8001/8002)。在 `application.yml` 中使用 `spring.profiles.active` 指定加载的配置文件,实现环境快速切换。本节还介绍了通过配置类读取参数的方法,适用于微服务场景,提升代码可维护性。课程源码可从 [Gitee](https://gitee.com/eson15/springboot_study) 下载。
842 0
|
Java 微服务 Spring
微服务——SpringBoot使用归纳——Spring Boot中的项目属性配置——少量配置信息的情形
在微服务架构中,随着业务复杂度增加,项目可能需要调用多个微服务。为避免使用`@Value`注解逐一引入配置的繁琐,可通过定义配置类(如`MicroServiceUrl`)并结合`@ConfigurationProperties`注解实现批量管理。此方法需在配置文件中设置微服务地址(如订单、用户、购物车服务),并通过`@Component`将配置类纳入Spring容器。最后,在Controller中通过`@Resource`注入配置类即可便捷使用,提升代码可维护性。
341 0
|
消息中间件 Java 数据库
新版 Seata 集成 RocketMQ事务消息,越来越 牛X 了!阿里的 Seata , yyds !
这里 借助 Seata 集成 RocketMQ 事务消息的 新功能,介绍一下一个新遇到的面试题:如果如何实现 **强弱一致性 结合**的分布式事务?
新版 Seata 集成 RocketMQ事务消息,越来越 牛X 了!阿里的 Seata , yyds !
|
消息中间件 数据管理 Serverless
阿里云消息队列 Apache RocketMQ 创新论文入选顶会 ACM FSE 2025
阿里云消息团队基于 Apache RocketMQ 构建 Serverless 消息系统,适配多种主流消息协议(如 RabbitMQ、MQTT 和 Kafka),成功解决了传统中间件在可伸缩性、成本及元数据管理等方面的难题,并据此实现 ApsaraMQ 全系列产品 Serverless 化,助力企业提效降本。

热门文章

最新文章