RocketMq基础认知

Posted 2023-04-23

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RocketMq基础认知相关的知识，希望对你有一定的参考价值。

参考技术A RocketMQ是一款分布式、队列模型的消息中间件，单机支持1万以上的持久化队列，前提是足够的内存、硬盘空间。

消息队列主要的应用场景：异步处理，应用解耦，流量削峰，消息通讯。

如上图所示， RocketMQ的部署结构有以下特点：

Name Server是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有Name Server。

Producer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从Name Server取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。

Consumer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

消息的生产者，负责发送消息，将消息推送给broker。一般由业务系统负责产生消息。

消息有3种发送方式：同步、异步、单向。

rocketmq的核心组件，负责消息的接收、存储（持久化到磁盘）、被消费者拉取消息等功能。

broker也存储消息相关的元数据，包括：消费者组、消费进度、topic&queue信息等。

消息的消费者，从broker上拉取消息从而进行消费。rocketmq提供两种消费者。

一般是后台系统负责异步消费消息。

主动消费者：DefaultMQ Pull Consumer，从broker中拉取一批消息并消费，主动权由消费者控制。

被动消费者：DefaultMQ Push Consumer，消费者实现回调接口，一旦有消息，broker回调接口，消费者被动响应。

注册中心的作用，提供轻量级的服务发现和提供路由信息（broker的服务注册与发现）。

nameserver存有全量的路由信息，提供对等的读写服务，支持快速扩缩容。

nameserver接收broker的请求，注册broker的路由信息。

nameserver接收client（producer/consumer）的请求，根据消息的topic获取相应的broker路由信息。

（手动创建的topic可以指定broker，自动创建的topic会随机指定broker，也许指定单个或全部，topic的概念在后面。）

一种消息的逻辑分类（消息的类型），比如说你有订单类的消息，也有库存类的消息，那么就需要进行分类存储。

生产者方面：发消息时需指定topic，可以有1-n个生产者发布1个topic的消息，

也1个生产者可以发布不同topic的消息。消费者方面：收消息时需订阅topic，

可以有1-n个消费者组订阅1个topic的消息，1个消费者组可以订阅不同topic的消息。

1个消息必须指定1个topic，topic允许自动创建与手工创建，topic创建时需要指定broker，可以指定1个或多个，

name server就是通过broker与topic的映射关系来做路由。

producer和consumer在生产和消费消息时，都需要指定消息的 topic，当topic匹配时，

consumer 才会消费到producer发送的消息。

topic与broker是多对多的关系，一个topic分布在多个broker上，一个broker可以配置多个topic。

message是消息的载体。每个message必须指定一个topic，相当于寄信的地址。

message还有一个可选的tag设置，以便消费端可以基于tag进行过滤消息。

message还有扩展的kv结构，例如你可以设置一个业务key到你的消息中，在broker上查找消息并诊断问题。

标签可以被认为是对topic的进一步细化。一般在相同业务模块中通过引入标签来标记不同用途的消息。

区分相同topic下不同种类的消息。

生产到哪个topic的哪个tag下，消费者也是从topic的哪个tag进行消费，即实现消息的过滤。

queue是消息的物理管理单位，而topic是逻辑管理单位。一个topic下可以有多个queue，

默认自动创建是4个，手动创建是8个。

queue的引入使得消息存储可以分布式集群化，具有了水平扩展的能力。

1个message只能属于1个queue、1个topic。

在rocketmq中，所有消息队列都是持久化，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，

访问其中的存储单元使用offset来访问，offset 为 java long 类型，64 位，理论上在 100年内不会溢出，

所以认为是长度无限，另外队列中只保存最近几天的数据，之前的数据会按照过期时间来删除。

也可以认为 Message Queue是一个长度无限的数组，offset就是下标。

rocketmq中，producer将消息发送给broker时，需要指定发送到哪一个queue中，默认情况下，

producer会轮询的将消息发送到每个queue中，顺序是随机的，但总体上每个queue的消息数量均分，

所有broker下的queue合并成一个list去轮询，

也可以由程序员通过MessageQueueSelector接口来指定具体发送到哪个queue中。

对于consumer而言，会为每个consumer分配固定的队列（如果队列总数没有发生变化），

consumer从固定的队列中去拉取没有消费的消息进行处理。

消费端会通过RebalanceService线程，10秒钟做一次基于topic下的所有队列负载，获取同一个Consumer Group下的所有Consumer实例数或Topic的queue的个数是否改变，通知所有Consumer实例重新做一次负载均衡算法。

理解成消费进度，可自增。

虽然每个topic下面有很多message queue，但是message queue本身并不存储消息。

真正的消息存储会写在CommitLog的文件，message queue只是存储CommitLog中对应的位置信息，

方便通过message queue找到对应存储在CommitLog的消息。不同的topic，

message queue都是写到相同的CommitLog 文件，也就是说CommitLog完全的顺序写。

服务启动顺序：name server->broker->producer&consumer

每个broker与name server集群中的所有节点建立长连接，

定时注册topic&broker的路由信息到所有name server中。

producer与name server集群中的其中一个节点（随机选择）建立长连接，

定期从name server获取topic路由信息，并向提供topic服务的broker master建立长连接，

且定期向broker master发送心跳，produce无状态，可集群部署。

producer只能将消息发送到broker master，但是consumer则不一样。

consumer与name server集群中的其中一个节点（随机选择）建立长连接，定期从name server获取topic路由信息，

consumer同时与提供topic服务的master和slave建立长连接且定时发送心跳，

consumer既可以从broker master订阅消息，也可以从broker slave订阅消息，订阅规则由broker配置决定。

broker一旦需要横向扩展，只需要启动更多的broker即可，然后把对应的topic建上，

客户端的queue集合即会变大，并且由于每个group下面的topic的配置都是独立的，

也就说可以让broker1下面的那个topic的queue数量是4，其他broker下的topic queue数量是2，

这样broker1则得到更大的负载。

rocketMQ基础环境

一、安装：

http://rocketmq.apache.org/dowloading/releases/

https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.5.1/rocketmq-all-4.5.1-bin-release.zip

启动

1.3.2 启动消患队列服务

启动单机的消息队列服务比较简单，不需要写配置文件，只需要依次启动

本机的NameServer 和B roker 即可。

启动NameServer:

> nohup sh bin/mqnamesrv &

> tail - f ~/Logs / rocketmqLogs / namesrv . Log

The Name Server boot success . ..

启动B roker :

> nohup sh bin/mqbroker n localhost : 9876&

> tail -f ~/ Logs/rocketmqLogs/broker . Log

The broker[%s, 192.168.0.233 : 10911] boot success .. .

1.3.3 用命令行发送和接收消息

为了快速展示发送和接收消息，本节展示的是用命令行发送和接收消息，

实际上就是运行写好的demo 程序，后续我们可以参考这些demo 来写自己的发

送和接收程序。

运行示例程序，发送和接收消息：

>export NAMESRV ADDR=localhost : 9876

> sh bin/tools . sh org .apache .roc ketmq.example . quickstart.Producer

SendResult [sendStatus=SEND OK, msgid=

> sh bin/tools.sh org .apache.rocketmq . example.quickstart . Consumer

ConsumeMessageThread 主d Receive New Messages : [MessageExt . ..

1.3.4 关闭消息队列

消息队列被启动后，如果不主动关闭，则会一直在后台运行，占用系统资

源。我们有专门用来关闭NameServ er 和Broker 的命令。

关闭NameServer 和Broker:

> sh bin/mqshutdown broker

The mqbroker (36695) i s running ...

Send shutdown request to mqbroker (36695 ) OK

> sh bin/mqshutdown namesrv

The mqnames r v (36664 ) is runn 工口g ...

Send shutdown request t o mqnamesrv (36664) OK

恭喜，现在你已经能够使用RocketMQ 发送并接收消息了，使用消息队列

的基本功能就是这么简单。