消息中间件——RocketMQ(一)
前言:此篇文章系本人学习过程中记录下来的笔记,里面难免会有不少欠缺的地方,诚心期待大家多多给予指教。
- 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 是阿里巴巴开源的一款分布式消息中间件产品,最初在阿里巴巴内部大规模使用,承担了海量消息的处理任务,如电商交易中的订单消息、物流消息等关键业务场景。2016 年,RocketMQ 正式捐赠给 Apache 软件基金会,并于 2017 年成为 Apache 顶级项目。它设计目标明确,旨在构建一个高性能、高可用、高可靠且功能丰富的消息中间件,以满足企业级分布式系统在不同业务场景下对消息通信的严格要求。
六、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 -Dmaven.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:努力到底,让持续学习成为贯穿一生的坚守。学习笔记持续更新中。。。。