当前位置：首页 > news >正文

消息中间件——RocketMQ（一）

news 来源：原创 2025/9/6 21:56:59

前言：此篇文章系本人学习过程中记录下来的笔记，里面难免会有不少欠缺的地方，诚心期待大家多多给予指教。

RocketMQ（一）

一、MQ出现的背景

在传统的单体应用架构中，系统的各个模块紧密耦合在一起。随着业务的快速发展和用户量的急剧增长，单体应用逐渐暴露出其局限性，例如可扩展性差、维护成本高、可靠性低等问题。于是，分布式系统架构应运而生，它将一个大型的应用拆分成多个小型的服务，每个服务都可以独立开发、部署和扩展，大大提高了系统的灵活性和可维护性。

然而，分布式系统也带来了新的挑战。不同服务之间需要进行高效、可靠的通信与数据交互。在这种情况下，消息中间件就发挥了至关重要的作用。消息中间件作为一种分布式系统中的基础设施，它为不同的应用程序提供了异步通信的能力。通过消息中间件，发送方可以将消息发送到消息队列中，而接收方可以根据自己的节奏从队列中获取消息进行处理，这样就实现了发送方和接收方的解耦，提高了系统的整体性能和可靠性。

二、MQ的主要产品

（一）、RabbitMQ

RabbitMQ 是一个开源的消息代理和队列服务器，它支持多种消息协议，如 AMQP、STOMP、MQTT 等。RabbitMQ 具有高可靠性、灵活性和可扩展性，它采用了 Erlang 语言编写，天生具备分布式和高并发的特性。RabbitMQ 在企业级应用中广泛使用，尤其在对可靠性要求极高的场景中表现出色。

（二）、Kafka

Kafka 是由 Apache 软件基金会开发的一个分布式流平台，最初由 LinkedIn 公司开发并开源。Kafka 以其高吞吐量、可持久化、分布式等特性而闻名。它主要应用于大数据领域，常用于日志收集、实时数据处理、流计算等场景。Kafka 采用了分区和副本机制，能够在大规模集群环境下高效地处理海量数据，并且保证数据的一致性和可靠性。

（三）、ActiveMQ

ActiveMQ 是使用 Java 语言开发的一款 MQ 产品，在早期发挥了重要作用，被众多公司和项目广泛应用。它支持 OpenWire、STOMP、AMQP 等多种消息协议，具备跨平台以及支持多种语言客户端的显著优点，为不同环境下的系统通信提供了便利。

然而，随着技术的飞速发展，ActiveMQ 逐渐暴露出一些问题。在高并发、大数据量处理等方面，它开始出现性能瓶颈。同时，其社区活跃度如今已变得很低，这意味着更新迭代的速度放缓，难以快速响应新的技术需求和解决潜在的漏洞。基于这些因素，在当下新开展的项目中，ActiveMQ 的使用已经很少见，众多开发者更倾向于选择性能更优、社区更活跃的消息中间件产品。

（四）、RocketMQ（本次学习）

RocketMQ 是阿里巴巴开源的一款分布式消息中间件产品，最初在阿里巴巴内部大规模使用，承担了海量消息的处理任务，如电商交易中的订单消息、物流消息等关键业务场景。2016 年，RocketMQ 正式捐赠给 Apache 软件基金会，并于 2017 年成为 Apache 顶级项目。它设计目标明确，旨在构建一个高性能、高可用、高可靠且功能丰富的消息中间件，以满足企业级分布式系统在不同业务场景下对消息通信的严格要求。

三、MQ的功能

总结有其如下三点：

（一）、限流削峰

MQ可以将系统的超量请求暂存其中，以便系统后期可以慢慢进行处理，从而避免了请求的丢失或系统被压垮。

（二）、异步解耦

上游系统对下游系统的调用若为同步调用，则会大大降低系统的吞吐量与并发度，且系统耦合度太高。而异步调用则会解决这些问题。所以两层之间若要实现由同步到异步的转化，一般性做法就是，在这两层间添加一个MQ层。

（三）、数据收集

分布式系统会产生海量级数据流，如：业务日志、监控数据、用户行为等。针对这些数据流进行实时或批量采集汇总，然后对这些数据流进行大数据分析，这是当前互联网平台的必备技术。通过MQ完成此类数据收集是最好的选择。

四、MQ的常见协议

（一）OpenWire

OpenWire 是 ActiveMQ 默认的消息传输协议，它是一种二进制协议，具有高效、轻量级的特点。

优点：提供了丰富的功能，如消息的持久化、事务支持、异步传输等。它能够在不同的平台和语言之间进行高效的消息传递，并且对 Java 语言有很好的支持。

缺点：由于其是二进制协议，可读性相对较差，在跨语言和跨平台的通用性上不如一些文本协议。

（二）、STOMP（Simple Text Oriented Messaging Protocol）

简单的面向文本的消息协议，它基于文本格式进行消息传输，具有良好的可读性和可扩展性。

优点：STOMP 协议非常灵活，易于理解和实现，支持多种编程语言和平台。它适用于各种消息中间件，能够方便地与不同的系统进行集成。

缺点：但由于是文本协议，在传输效率上相对二进制协议可能会稍低一些。

（三）、AMQP（Advanced Message Queuing Protocol）

高级消息队列协议，是一个面向消息、队列、路由（包括点对点和发布 / 订阅）的、可靠的、安全的、跨平台的应用层协议。

优点：AMQP 具有很强的规范性和通用性，提供了丰富的功能集，如消息确认、事务处理、消息优先级等。它能够在不同的消息中间件产品之间实现互操作性，适合在企业级分布式系统中使用。

缺点：AMQP 的复杂性相对较高，实现和配置较为复杂，对开发者的要求也比较高。

（四）、MQTT（Message Queuing Telemetry Transport）

消息队列遥测传输协议，是一种基于发布 / 订阅模式的轻量级消息协议，主要用于物联网（IoT）场景中的设备之间的通信。

优点：MQTT 具有低功耗、低带宽占用、支持大量客户端等特点，能够在资源受限的设备和网络环境中高效运行。它采用简单的消息格式和连接方式，适合于不稳定的网络连接。

缺点：由于其主要针对物联网场景，在一些复杂的企业级应用中可能功能不够丰富。

（五）、Kafka Protocol

Kafka 自己的消息协议，它是为了满足 Kafka 高吞吐量、低延迟的分布式消息系统的需求而设计的。

优点：Kafka 协议具有高效的二进制格式，能够支持大规模的消息处理和快速的数据传输。它在分布式环境下具有良好的扩展性和容错性，适用于处理海量的实时数据。

缺点：Kafka 协议相对较为特定于 Kafka 系统，与其他消息中间件的兼容性较差。

五、RocketMQ概述

六、RocketMQ的安装与使用

（一）、基本概念

1、Message（消息）

消息是指，消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。

2、Topic（主题）

Topic 是 RocketMQ 中的消息分类，用于区分不同类型的消息。生产者将消息发送到特定的 Topic，消费者通过订阅感兴趣的 Topic 来接收消息。一个 Topic 可以有多个生产者和多个消费者，类似于现实生活中的不同话题，不同的人可以围绕特定话题进行交流。

3、Tag（标签）

为消息设置的标签，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。

4、Queue（队列）

每个 Topic 可以划分为多个 Queue，Queue 是消息存储和负载均衡的基本单位。通过将消息分散存储在多个 Queue 中，RocketMQ 能够实现高并发的读写操作，提高系统的性能和吞吐量。消费者通过从不同的 Queue 拉取消息来进行消费，从而实现并行处理。

一个Topic的Queue中的消息只能被一个消费者组中的一个消费者消费。一个Queue中的消息不允许同一个消费者组中的多个消费者同时消费。

5、消息标识（MessageId/Key）

RocketMQ中每个消息拥有唯一的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。不过需要注意的是，MessageId有两个：在生产者send()消息时会自动生成一个MessageId（msgId)，当消息到达Broker后，Broker也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId与key都称为消息标识。

msgId：由producer端生成，其生成规则为： producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode + 当前时间 + AutomicInteger自增计数器。
offsetMsgId：由broker端生成，其生成规则为：brokerIp + 物理分区的offset（Queue中的偏移量）。
key：由用户指定的业务相关的唯一标识。

（二）、系统架构

1、Producer

消息生产者，负责生产消息。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

例如，业务系统产生的日志写入到MQ的过程，就是消息生产的过程

再如，电商平台中用户提交的秒杀请求写入到MQ的过程，就是消息生产的过程

RocketMQ中的消息生产者都是以生产者组（Producer Group）的形式出现的。生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。

2、Consumer

消息消费者，负责消费消息。一个消息消费者会从Broker服务器中获取到消息，并对消息进行相关业务处理。

例如，系统从MQ中读取日志，并对日志进行解析处理的过程就是消息消费的过程。

再如，电商平台的业务系统从MQ中读取到秒杀请求，并对请求进行处理的过程就是消息消费的过程。

RocketMQ中的消息消费者都是以消费者组（Consumer Group）的形式出现的。消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息。消费者组使得在消息消费方面，实现负载均衡（将一个Topic中的不同的Queue平均分配给同一个Consumer Group的不同的Consumer，注意，并不是将消息负载均衡）和容错（一个Consmer挂了，该Consumer Group中的其它Consumer可以接着消费原Consumer消费的Queue）的目标变得非常容易。

消费者组中Consumer的数量应该小于等于订阅Topic的Queue数量。如果超出Queue数量，则多出的 Consumer将不能消费消息。

不过，一个Topic类型的消息可以被多个消费者组同时消费。注意：

1）消费者组只能消费一个Topic的消息，不能同时消费多个Topic消息

2）一个消费者组中的消费者必须订阅完全相同的Topic

3、Name Server

NameServer是一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现。 RocketMQ的思想来自于Kafka，而Kafka是依赖了Zookeeper的。所以，在RocketMQ的早期版本，即在MetaQ v1.0与v2.0版本中，也是依赖于Zookeeper的。从MetaQ v3.0，即RocketMQ开始去掉了Zookeeper依赖，使用了自己的NameServer。主要包括两个功能：

Broker管理：接受Broker集群的注册信息并且保存下来作为路由信息的基本数据；提供心跳检测机制，检查Broker是否还存活。

路由信息管理：每个NameServer中都保存着Broker集群的整个路由信息和用于客户端查询的队列信息。Producer和Conumser通过NameServer可以获取整个Broker集群的路由信息，从而进行消息的投递和消费。比如路由注册、发现、剔除等。

①、路由注册

NameServer通常也是以集群的方式部署，不过，NameServer是无状态的，即NameServer集群中的各个节点间是无差异的，各节点间相互不进行信息通讯。那各节点中的数据是如何进行数据同步的呢？在Broker节点启动时，轮询NameServer列表，与每个NameServer节点建立长连接，发起注册请求。在NameServer内部维护着⼀个Broker列表，用来动态存储Broker的信息。

注意，这是与其它像zk、Eureka、Nacos等注册中心不同的地方。

这种NameServer的无状态方式，有什么优缺点：

优点：NameServer集群搭建简单，扩容简单。

缺点：对于Broker，必须明确指出所有NameServer地址。否则未指出的将不会去注册。也正因为如此，NameServer并不能随便扩容。因为，若Broker不重新配置，新增的NameServer对于Broker来说是不可见的，其不会向这个NameServer进行注册。

Broker节点为了证明自己是活着的，为了维护与NameServer间的长连接，会将最新的信息以心跳包的方式上报给NameServer，每30秒发送一次心跳。心跳包中包含BrokerId、Broker地址(IP+Port)、 Broker名称、Broker所属集群名称等等。NameServer在接收到心跳包后，会更新心跳时间戳，记录这个Broker的最新存活时间。

②、路由剔除

由于Broker关机、宕机或网络抖动等原因，NameServer没有收到Broker的心跳，NameServer可能会将其从Broker列表中剔除。NameServer中有⼀个定时任务，每隔10秒就会扫描⼀次Broker表，查看每一个Broker的最新心跳时间戳距离当前时间是否超过120秒，如果超过，则会判定Broker失效，然后将其从Broker列表中剔除。

扩展：对于RocketMQ日常运维工作，例如Broker升级，需要停掉Broker的工作。OP（运维工程师）需要怎么做？

OP需要将Broker的读写权限禁掉。一旦client(Consumer或Producer)向broker发送请求，都会收到broker的NO_PERMISSION响应，然后client会进行对其它Broker的重试。当OP观察到这个Broker没有流量后，再关闭它，实现Broker从NameServer的移除。

③、路由发现

RocketMQ的路由发现采用的是Pull模型。当Topic路由信息出现变化时，NameServer不会主动推送给客户端，而是客户端定时拉取主题最新的路由。默认客户端每30秒会拉取一次最新的路由。

扩展：

1）Push模型：推送模型。其实时性较好，是一个“发布-订阅”模型，需要维护一个长连接。而长连接的维护是需要资源成本的。该模型适合于的场景：实时性要求较高 Client数量不多，Server数据变化较频繁。

2）Pull模型：拉取模型。存在的问题是，实时性较差。

3）Long Polling模型：长轮询模型。其是对Push与Pull模型的整合，充分利用了这两种模型的优势，屏蔽了它们的劣势。

④、客户端NameServer选择策略

这里的客户端指的是Producer与Consumer

客户端在配置时必须要写上NameServer集群的地址，那么客户端到底连接的是哪个NameServer节点呢？客户端首先会生产一个随机数，然后再与NameServer节点数量取模，此时得到的就是所要连接的节点索引，然后就会进行连接。如果连接失败，则会采用round-robin策略，逐个尝试着去连接其它节点。首先采用的是随机策略进行的选择，失败后采用的是轮询策略。

扩展：Zookeeper Client是如何选择Zookeeper Server的？

简单来说就是，经过两次Shuffle，然后选择第一台Zookeeper Server。

详细说就是，将配置文件中的zk server地址进行第一次shuffle，然后随机选择一个。这个选择出的一般都是一个hostname。然后获取到该hostname对应的所有ip，再对这些ip进行第二次shuffle，从shuffle过的结果中取第一个server地址进行连接。

4、Broker

Broker充当着消息中转角色，负责存储消息、转发消息。Broker在RocketMQ系统中负责接收并存储从生产者发送来的消息，同时为消费者的拉取请求作准备。Broker同时也存储着消息相关的元数据，包括消费者组消费进度偏移offset、主题、队列等。

Remoting Module：整个Broker的实体，负责处理来自clients端的请求。而这个Broker实体则由以下模块构成。

Client Manager：客户端管理器。负责接收、解析客户端(Producer/Consumer)请求，管理客户端。例如，维护Consumer的Topic订阅信息。

Store Service：存储服务。提供方便简单的API接口，处理消息存储到物理硬盘和消息查询功能。

HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。

Index Service：索引服务。根据特定的Message key，对投递到Broker的消息进行索引服务，同时也提供根据Message Key对消息进行快速查询的功能。

5、工作流程

启动NameServer，NameServer启动后开始监听端口，等待Broker、Producer、Consumer连接。
启动Broker，Broker会与所有的NameServer建立并保持长连接，然后每30秒向NameServer定时发送心跳包。
发送消息前，可以先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，当然，在创建Topic时也会将Topic与Broker的关系写入到NameServer中。不过，这步是可选的，也可以在发送消息时自动创建Topic。
Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取路由信息，即当前发送的Topic消息的Queue与Broker的地址（IP+Port）的映射关系。然后根据算法策略从队选择一个Queue，与队列所在的Broker建立长连接从而向Broker发消息。当然，在获取到路由信息后，Producer会首先将路由信息缓存到本地，再每30秒从NameServer更新一次路由信息。
Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取其所订阅Topic的路由信息，然后根据算法策略从路由信息中获取到其所要消费的Queue，然后直接跟Broker建立长连接，开始消费其中的消息。Consumer在获取到路由信息后，同样也会每30秒从NameServer更新一次路由信息。不过不同于Producer的是，Consumer还会向Broker发送心跳，以确保Broker的存活状态。

（三）、单机安装与启动

1、准备工作

①、环境配置

②、下载压缩包

wget https://archive.apache.org/dist/rocketmq/5.3.2/rocketmq-all-5.3.2-bin-release.zip

这种方式，太慢了，建议官网下载，然后上传到服务器上。下载 | RocketMQ

解压：

unzip zip文件 
###-d /path/to/directory 指定目录，不指定默认在当前目录

2、修改初始内存

①、修改runserver.sh

修改前：

修改后：

只要能保证MQ能正常启动就行。

②、修改runbroker.sh

修改前：

修改后：

3、启动

①、NameServer启动

$ nohup sh bin/mqnamesrv & #如果是这样启动，pwd应该是/opt/rocketmq-all-5.3.2-bin-release$ nohup sh ./mqnamesrv & #如果是这样启动，pwd应该是/opt/rocketmq-all-5.3.2-bin-release
/bin

验证是启动成功：

1、根据后台进程判断

2、或者根据启动日志查看

$ tail -f ~/logs/rocketmqlogs/namesrv.log

②、Broker启动

nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &

4、测试发送/接受消息

①、控制台测试

发送消息：

消费消息：

②、Java代码测试

新建普通Java项目，导入依赖。

        <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client-java</artifactId><version>5.0.5</version></dependency>

生产者代码：

public static void main(String[] args) throws ClientException, IOException {// 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081。String endpoint = "xxxx:8081";// 消息发送的目标Topic名称，需要提前创建。String topic = "TestTopic";ClientServiceProvider provider = ClientServiceProvider.loadService();ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);ClientConfiguration configuration = builder.build();// 初始化Producer时需要设置通信配置以及预绑定的Topic。Producer producer = provider.newProducerBuilder().setTopics(topic).setClientConfiguration(configuration).build();// 普通消息发送。Message message = provider.newMessageBuilder().setTopic(topic)// 设置消息索引键，可根据关键字精确查找某条消息。.setKeys("messageKey")// 设置消息Tag，用于消费端根据指定Tag过滤消息。.setTag("messageTag")// 消息体。.setBody(("java测试RocketMQ发送消息").getBytes()).build();try {// 发送消息，需要关注发送结果，并捕获失败等异常。SendReceipt sendReceipt = producer.send(message);System.out.println("Send message successfully, messageId={}" + sendReceipt.getMessageId());} catch (ClientException e) {System.out.println("Failed to send message" + e);}producer.close();}

消费者代码：

public static void main(String[] args) throws ClientException, IOException, InterruptedException {// 加载客户端服务提供者ClientServiceProvider provider = ClientServiceProvider.loadService();// 配置 NameServer 地址String endpoints = "xxxx:8081";ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder().setEndpoints(endpoints).build();// 消费组名称String consumerGroup = "YourConsumerGroup";// 要订阅的主题String topic = "TestTopic";// 创建过滤表达式，这里使用 * 表示不过滤FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);// 创建 PushConsumer 实例PushConsumer pushConsumer = provider.newPushConsumerBuilder().setClientConfiguration(clientConfiguration).setConsumerGroup(consumerGroup).setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)).setMessageListener(messageView -> {try {//打印出生产者产生的消息ByteBuffer bodyBuffer = messageView.getBody();byte[] body = new byte[bodyBuffer.remaining()];bodyBuffer.get(body);String messageContent = new String(body);System.out.println("Message content: " + messageContent);System.out.println("Consume message successfully, messageId=" + messageView.getMessageId());return ConsumeResult.SUCCESS;} catch (Exception e) {System.err.println("Error consuming message: " + e.getMessage());return ConsumeResult.FAILURE;}}).build();// 程序休眠一段时间，让消费者有时间接收消息Thread.sleep(Long.MAX_VALUE);// 关闭消费者pushConsumer.close();}

启动生产者，消费者，控制台输出：

5、关闭Server

先关broker，在关namesrv。

（四）、控制台的安装与启动

RocketMQ有一个可视化的dashboard，通过该控制台可以直观的查看到很多数据

1、下载

https://github.com/apache/rocketmq-externals/tags，下载完成后上传到Linux服务器(tar.gz)，本地windows服务(zip)

2、修改配置

3、添加依赖

        <dependency><groupId>javax.xml.bind</groupId><artifactId>jaxb-api</artifactId><version>2.3.0</version></dependency><dependency><groupId>com.sun.xml.bind</groupId><artifactId>jaxb-impl</artifactId><version>2.3.0</version></dependency><dependency><groupId>com.sun.xml.bind</groupId><artifactId>jaxb-core</artifactId><version>2.3.0</version></dependency><dependency><groupId>javax.activation</groupId><artifactId>activation</artifactId><version>1.1.1</version></dependency>

修改SpringBoot 版本为2.7.12，以及对应的jdk版本：

4、打包

4.1、安装maven(有直接跳过)

①、下载maven

wget https://dlcdn.apache.org/maven/maven-3/3.9.6/binaries/apache-maven-3.9.6-bin.tar.gz

②、解压

tar -zxvf apache-maven-3.9.6-bin.tar.gz

③、配置环境变量

maven.sh不存在，vim会自动创建，保存退出。

④、配置生效

source /etc/profile.d/maven.sh

④、验证maven版本

mvn -version

4.2、mvn打包

mvn clean package -Dmave.test.skip=true

4.3、异常情况

上面的打包可能会失败，由于JVM内存不足导致。在当前的终端会话中临时设置了 MAVEN_OPTS 环境变量。

export MAVEN_OPTS="-Xmx2048m -Xms1024m"

删除target目录下的临时文件，然后重新运行 Maven 构建：

rm -rf target/ 
mvn clean package -Dmaven.test.skip=true

5、测试访问

启动项目

访问服务器地址加端口号：

可视化界面可比控制台好看多了。。。

五、总结

消息中间件在当今的分布式系统中已经成为不可或缺的一部分，它有效地解决了分布式系统中不同服务之间的通信和数据交互问题。无论是在互联网企业的大型分布式系统中，还是在传统企业的信息化建设中，RocketMQ 都能够发挥其强大的优势，为系统的稳定运行和业务的高效发展提供有力支持。通过本文对 RocketMQ 的学习，能够实际项目中根据需求合理选择和使用消息中间件。

ps：努力到底，让持续学习成为贯穿一生的坚守。学习笔记持续更新中。。。。