RocketMQ 必须知道的理论知识

2023-03-31
1 min read

RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。

简介

项目地址：https://github.com/apache/rocketmq/tree/master

RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。ConsumerGroup 由多个Consumer 实例构成。
img

消息的发布是指某个生产者向某个topic发送消息；消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息，进而从该topic消费数据。

架构

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

  • Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。 注意：当前RocketMQ版本在部署架构上支持一Master多Slave，但只有BrokerId=1的从服务器才会参与消息的读负载。

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

  • Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从Master还是Slave拉取。

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包含了以下几个重要子模块。

    • Remoting Module：整个Broker的实体，负责处理来自Client端的请求。
    • Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息。
    • Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
    • HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
    • Index Service：根据特定的Message key对投递到Broker的消息进行索引服务，以提供消息的快速查询。

部署

结合部署架构图，描述集群工作流程：

  • 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
  • Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
  • 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
  • Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
  • Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

单节点部署

设置环境变量

1  set -x ROCKETMQ_HOME        /Users/van/soft/rocketMQ

下载项目至ROCKETMQ_HOME文件目录，
https://github.com/apache/rocketmq/tree/master
下载项目至ROCKETMQ_HOME文件目录，
https://github.com/apache/rocketmq-dashboard

启动NameServer

1  nohup sh $ROCKETMQ_HOME/rocketmq-4.9.3/bin/mqnamesrv &
2  tail -f ~/logs/rocketmqlogs/namesrv.log
3  jps

启动borkerServer

1  nohup sh $ROCKETMQ_HOME/rocketmq-4.9.3/bin/mqbroker -n localhost:9876 &
2  tail -f ~/logs/rocketmqlogs/broker.log
3  jps

启动管理端

项目地址：https://github.com/apache/rocketmq-dashboard

修改nameserver地址，application.yml

1      ...
2      namesrvAddrs:
3        - 127.0.0.1:9876
4      ...

启动命令

mvn springboot:run

访问地址：http://localhost:8080/#/topic

消息发送和消费测试

1  #bash
2  export NAMESRV_ADDR=localhost:9876
3  #fish
4  set -x NAMESRV_ADDR localhost:9876
5  sh $ROCKETMQ_HOME/rocketmq-4.9.3/bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
6  sh $ROCKETMQ_HOME/rocketmq-4.9.3/bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

复制和刷盘策略

复制策略：同步复制、异步复制，broker master与 slaver 之间数据同步策略
刷盘策略：同步刷盘、异步刷盘，异步刷盘一般是将数据存储至pageCache,达到一定数量时自动进行落盘

blocker 集群模式

单Master

只有一个blocker，只能在测试时使用，有单点故障

多Master

broker仅由多个master构成，不存在slaver。同一个topic的各个queue会平均分布至多个master节点。
优点：配置简单、单个master宕机对集群没影响。前提是需要配置磁盘阵列。raid磁盘阵列的效率要高于master-slaver集群，raid是硬件支持，成本较高。
缺点：未恢复之前，该机器上的消息不能被消费，消息时时性会受到影响。

多Master多Slaver-异步复制

master与slaver之间是主备关系，即master负责消息的读写请求，而slaver仅负责消息的备份，mater宕机后的角色自动切换。
由于是异步负责，在切换过程中可能会涉及少量消息丢失。取决于master向slaver同步数据的时机。

多Master多Slaver-同步双写

消息写入master后，等待master将信息同步至slaver成功后，返回成功。
优点：不存在丢失
缺点：单个消息的RT高，导致性能要略低（约10%），master宕机后不会自动切换至slaver。

RocketMq工作原理

消息的生产过程

  • Producer发送消息之前，会先向nameserver请求消息topic的路由信息
  • nameserver返回该topic的路由表和broker列表
    路由表：map，key为topic名称、value是一个queueData实例列表，即只要涉及到该topic的broker，一个broker对应一个queue,Queuedata中包含brokerName。简单来说，路由表的value为所有涉及该topic的brokerName
    routerMap<topicName,List>,queueData: brokerName
    Borker列表：Map<brokerName、brokerDataMap>, brokerDataMap<brokerId,brokerIp>
  • Producer根据消息的选择策略，从queue列表中选出一个队列，用于后续存储消息
  • Producer对消息做一些特殊出理，例如消息超过4M，对消息进行压缩
  • producer向选择出的queue所在的broker发出rpc请求，将消息发送到选择出的queue

queue选择算法

  • 轮询算法，每个queue可均匀收到消息，由于需要等到投递成功后才能进行下个节点的轮询，因此该算法存在在生产者端消息积压问题，影响消息的投递性能。
  • 最小投递延迟算法，将消息投递至延迟最小的queue,可有效提升消息的投递性能。问题是可能出现单机过热，消息分配不均。

消息存储

  • 数据存储在home/store目录下，该目录在启动时创建，正常关闭broker该消息会自动消失，如果启动前发现此文件存在，则说明之前是非正常关闭。
  • 其中存储着commitlog文件，而消息是写在commitlog文件中的
  • config目录，存放着broker运行期间的一些配置数据
  • consumequeue, 存储着消费队列
  • index存储着消息索引文件
  • lock:运行期间使用到的全局资源锁

commitlog文件

  • 偏移量
  • 索引文件

Reblance

没次新增或者删除消费者的时候都会触发reblance，reblance会导致消息积压和消息重复消费的问题。

另外如果消费者数量大于queue时，消息只会负载至有限的消费者，这个叫reblance的限制

#Java
Next 推荐一款 org mode 导出 html 的样式

See Also

java 栈溢出问题排查
H2 Usage
Ftp Resume Transfer
Maven 项目骨架
Clickhouse Tutorial
ElasticSearch Part1