必须先理解的RocketMQ入门手册，才能再次深刻解读

RocketMQ入门手册

RocketMQ是一个分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具备高性能、高可靠、高实时、分布式特色，

同时，普遍应用于多个领域，包括异步通讯解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通讯、移动应用、手游、视频、物联网、车联网等。

具备如下特色：

• 可以保证严格的消息顺序
• 提供丰富的消息拉取模式
• 高效的订阅者水平扩展能力
• 实时的消息订阅机制
• 亿级消息堆积能力

RocketMQ 架构原理分析

RocketMQ 架构

NameServer （名称服务器）：

• 提供轻量级的服务发现和路由。NameServer接受来自Broker群集的注册，并提供检测信号机制以检查Broker是否还存在
• 每一个NameServer记录完整的路由信息（Broker 相关 Topic 等元信息，并给 Producer 提供 Consumer 查找 Broker 信息），提供相应的读写服务。

Broker（消息服务器）： 消息存储中心，接收来自 Producer 的消息并存储， Consumer 从这里取得消息

• 单个Broker节点与全部的NameServer节点保持长链接及心跳，并会定时将Topic信息注册到NameServer，（其底层通讯是基于Netty实现的）
• Broker负责消息存储，以Topic为维度支持轻量级的队列，单机能够支撑上万队列规模，支持消息推拉模型。
• 具备上亿级消息堆积能力，同时可严格保证消息的有序性

Producer （生产者）：

• 负责产生消息，生产者向消息服务器发送由业务应用程序系统生成的消息
• 生产者支持分布式部署。 分布式生产者经过多种负载平衡模式将消息发送到Broker集群。 发送过程支持快速失败而且延迟低
• 三种方式发送消息：同步、异步和单向

Consumer（消费者）：

• 负责消费消息，消费者从消息服务器拉取信息并将其输入用户应用程序
• 也支持“推和拉”模型中的分布式部署。
• 它还支持集群使用和消息广播。 它提供了实时消息订阅机制，能够知足大多数消费者的需求。

Broker Server

Broker Server负责消息的存储和传递，消息查询，HA高可用等，Broker Server几个主要模块组成：

Remoting Module（远程模块）：broker入口，处理来自客户端的请求

Client Manager（客户端管理）：管理client（生产者/消费者）并维护消费者的主题订阅

Store Service（存储服务）：提供简单的API中数据库中存储或查询消息

HA Service（高可用服务）：提供master broker和slave broker之间的数据同步功能

Index Service（索引服务）：将message创建索引来提供快速的查询能力

RocketMQ 总体流程

1. 启动 NameServer，NameServer启动后进行端口监听，等待 Broker、Producer、Consumer 连上来，至关于一个路由控制中心

2. Broker 启动，跟全部的 Namesrv 保持长链接，定时发送心跳包

   • 心跳包中，包含当前 Broker 信息(IP+端口等)以及存储全部 Topic 信息
   • 注册成功后，Namesrv 集群中就有 Topic 跟 Broker 的映射关系
3. 收发消息前，先建立 Topic 。建立 Topic 时，须要指定该 Topic 要存储在哪些 Broker上。也能够在发送消息时自动建立Topic

4. Producer 发送消息

   • 启动时，先跟 Namesrv 集群中的其中一台创建长链接，并从Namesrv 中获取当前发送的 Topic 存在哪些 Broker 上
   • 而后跟对应的 Broker 创建长链接，直接向 Broker 发消息

5. Consumer 消费消息

   • 跟其中一台 Namesrv 创建长链接，获取当前订阅 Topic 存在哪些 Broker 上
   • 而后直接跟 Broker 创建链接通道，开始消费消息*RocketMQ的消息领域模型

RocketMQ Message

Topic（主题）： 表示消息的第一级类型，是最细粒度的订阅单位（生产者传递消息和消费者提取消息标识）

• 一条消息必须有一个Topic
• 一个Group能够订阅多个Topic的消息
• Topic通常为领域范围，好比交易消息

Tag（标签）： 表示消息的第二级类型，能够是使用相同的Topic不一样的Tag来表示同一业务模块的不一样任务的消息，好比交易消息又能够分为：交易建立消息，交易完成消息等

• 助于保持代码整洁和一致
• 简化RocketMQ提供的查询系统

Message（消息体）： 消息是要传递的信息。 Message中必须包含一个Topic，可选Tag和key-vaule键值对

Message Queue（消息队列）： 全部消息队列都是持久化

• 一个Topic下能够有多个Queue
• Queue的引入使得消息的存储能够分布式集群化，具备了水平扩展能力

Group（组）： 分为Producer Group（生产者组）和Consumer Group（消费者组），具备相同角色组成Group

• 原生产者在交易后崩溃，broker能够联系同一辈子产者组的不一样生产者实例以进行提交或回退交易。
• 消费者组的消费者实例必须具备彻底相同的主题订阅

RocketMQ 特性

Message Model（消息模式）：

• Clustering（集群式）：当使用集群消费模式时，MQ 认为任意一条消息只须要被集群内的任意一个消费者处理便可
• Broadcasting（广播式）：当使用广播消费模式时，MQ 会将每条消息推送给集群内全部注册过的客户端，保证消息至少被每台机器消费一次

Message Order（消息顺序）

• 使用DefaultMQPushConsumer时，能够决定按顺序或同时使用消息

  • Orderly：有序地使用消息意味着消息的消费顺序与生产者为每一个消息队列发送消息的顺序相同。（ 若是要处理必须强制执行全局顺序的状况，请确保您使用的主题只有一个消息队列）

  若是指定按顺序使用，则消息使用的最大并发度是使用者组订阅的消息队列数

  • Concurrently：同时使用消息时，消息使用的最大并发性仅受为每一个使用方客户端指定的线程池限制

  在此模式下再也不保证消息顺序

Message Types（消息类型）

• 事务消息
• 顺序消息
• 延迟消息

RocketMQ单机版安装

1. 下载编译源码

   # 下载$ 
     > wget wget http://mirror.bit.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-source- > 
     # 解压$
     >unzip rocketmq-all-4.7.0-source-release.zip
     > cd rocketmq-all-4.7.0/
     # 编译$
     > mvn -Prelease-all -DskipTests clean install -U
     > cd distribution/target/rocketmq-4.7.0/rocketmq-4.7.0

2. 启动 Name Server

   # 启动 Name Server 服务
    > nohup sh bin/mqnamesrv &
    # 启动完成后，查看日志$
    > tail -f ~/logs/rocketmqlogs/namesrv.log
     The Name Server boot success...

3. 启动 Broker

   在 conf 目录下，RocketMQ 提供了多种 Broker 的配置文件：

   • broker.conf ：单主，异步刷盘。
   • 2m/ ：双主，异步刷盘。
   • 2m-2s-async/ ：两主两从，异步复制，异步刷盘。
   • 2m-2s-sync/ ：两主两从，同步复制，异步刷盘。
   • dledger/ ：Dledger 集群，至少三节点

   # 启动 Broker服务
    > nohup sh bin/mqbroker -n localhost:9876 &
    # 启动完成后，查看日志$
    > tail -f ~/logs/rocketmqlogs/broker.log 
     The broker[%s, 172.30.30.233:10911] boot success...

其中，参数：

• 经过 -c 参数，配置读取的主 Broker 配置
• 经过 -n 参数，设置 RocketMQ Namesrv 地址

1. Send & Receive Messages（消息发送与接收）

   在发送/接收消息以前，咱们须要告知client（生产者/消费者）Name Servers的地址。 RocketMQ提供了多种方法来实现：

   • 在代码中设置：producer.setNamesrvAddr("ip:port")
   • java属性配置：rocketmq.namesrv.addr
   • 环境变量配置：NAMESRV_ADDR
   • HTTP Endpoint

为简单起见，咱们使用环境变量：NAMESRV_ADDR，以下所示：

# 设置 Name Servers的地址$
 > export NAMESRV_ADDR=localhost:9876
 # 生产消息$
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
 SendResult [sendStatus=SEND_OK, msgId= ...
 # 消费消息$ 
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
 ConsumeMessageThread_%d Receive New Messages: [MessageExt...

各位看官还能够吗？喜欢的话，动动手指点个💗，点个关注呗！！谢谢支持！

欢迎关注公众号【Ccww技术博客】，原创技术文章第一时间推出