RocketMQ（一）：消息中央件缘起，一览整体架构及核心组件 ...

RocketMQ（一）：消息中央件缘起，一览整体架构及核心组件

消息队列MessageQueue，简称MQ
在队列的基础上，参加生产者与消耗者模型，使用队列作为载体就可以大概组成简单的消息队列，在队列中“运输”的数据被称为消息

消息队列可以在单节点内存中使用，也可以作为分布式存储的中央件来使用
由于项目的架构组织，现在常接触的消息队列往往是作为分布式存储的消息中央件来使用，比如：RabbitMQ、RocketMQ、Kafka等
内存队列相比于消息中央件往往有轻量、低延迟（无需网络通信）、简单易用的特点，但也存在不能持久化（消息丢了怎么办？）、无法扩展（消息量太大怎么办？）的缺陷
而消息中央件的特点较多如：持久化、高可用、集群扩展、负载均衡、系统解耦等特点，但同时也会增加调用链路、提升系统复杂度，因此常用于分布式系统中
   特点
  异步通信：MQ提供异步通信，无需同步等待，得当必要异步场景
持久化：消息会进行持久化，持久化后无需担心异步通信的消息会丢失
削峰填谷：面对突发流量，MQ相当于缓冲区，防止后端服务短时间内接收过多请求导致服务崩溃
系统解耦：松耦合，生产者（调用方）、消耗者（被调用方）可以独立升级/扩展
集群：与其他中央件集群类似，方便水平/垂直扩展，提高系统吞吐量/可用性
消息中央件除了这些特点外，另有它们独有的功能与特点，本文就从RocketMQ开始，快速入门消息中央件专栏
在专栏中一步步分析消息中央件的架构、流程、原理、源码等，再分析各种消息中央件的上风以及适用场景
RocketMQ架构概念

• Message：消息为MQ中运输的载体，存储必要传输的数据与其他元数据
  
• MessageQueue：消息队列用于存储消息，内部通过偏移量可以大概找到消息
  
  
  
  • 分为读写队列用于消耗时读和持久化消息时写，通常队列数量相同
    
  • 队列ID使用数量0开始并渐渐进行自增，比如分配3个读写队列，那么id分别为0、1、2
    
  
  
• Topic：主题（类似Kafka中的分区），生产者必要将消息发送到对应的Topic上，消耗者必要订阅对应的Topic进行消耗
  
  
  
  • 充当消息的分类，过滤消息，比如不同业务(量级)的消息分发到对应的Topic中（order、pay、cart、user…）
    
  • Topic中存在多个队列（MQ）用于存储消息，增加topic下的队列可以大概提高消息的水平写入本领
    
  • Topic可以存在不同的broker上，保证高可用
    
  
  
• Tag：主题下的二级分类，过滤消息
  
• Broker：存储消息、MessageQueue、Topic等元数据的服务端，用于接收消息（处理惩罚生产）、持久化消息、查找消息（处理惩罚消耗）
  
  
  
  • 接收消息只能由主节点接收并持久化，从节点只用于同步对应主节点消息
    
  • 消耗消息既可以通过主节点拉取消息，也可以通过从节点
    
  
  

通过下面两个主节点的Broker图，很容易的可以明白它们的关系：

• Broker为存储消息的服务端，其中包含多个Topic
  
• Topic是用于生产、消耗订阅的主题，其中为了可以大概水平扩展写入性能可以设置多个MessageQueue，MessageQueue的ID从0开始进行自增
  
• 为了保证高可用，不同的Broker也会存在相同的Topic，只是其中的队列不同，防止broker意外宕机时服务不可用，如图中的TopicA，0、1队列在Broker A中，而2、3队列在Broker B中
  

• NameServer：存储broker路由信息、类似注册中央
  
  
  
  • 与broker通信，存储其数据如：Topic、MessageQueue等数据，顺便进行心跳判断broker是否离线
    
  • 与producer、consumer通信，将broker元数据进行传输，这样无论是生产照旧消耗都能根据数据找到对应的Broker、Topic、MQ等
    
  
  
• Product：生产者，用于生产消息，并把消息发送到消息队列
  
  
  
  • 相同设置的生产者成组Group可以和谐工作
    
  • 通过NameServer通信获取到的路由信息，根据负载均衡算法选择对应的Topic以及队列ID发送到对应Broker中进行持久化
    
  
  
• Comsumer：消耗者，用于消耗消息，从消息队列拉取消息(长轮询)进行消耗消息
  
  
  
  • 队列中使用偏移量确认消息，消耗时同理也使用偏移量标识消耗到的位置
    
  • 消耗模式分为广播、集群模式，广播模式就是发布订阅模型，集群模式为点对点消耗
    
  • 拉取消息使用长轮询机制弥补实时性差的特点，但大量长连接会导致开销大（后文详细描述长轮询机制）
    
  • 通过NameServer通信获取到的路由信息，消耗者根据消耗模式（广播/集群）选择对应的Topic，根据推送/拉取的方式获取消息
    
  • Group 同组消耗者和谐工作均衡消耗消息，集群模式下一个队列最多对应一个消耗者，如果消耗者数量凌驾队列数量则无效
    
  
  

通过以下的架构图，可以大概容易明白NameServer、Broker、Product、Consumer集群之间的关系：

• NameServer集群启动
  
• Broker集群设置NameServer集群地址并启动，向每个NameServer节点心跳的同时携带自身broker中的Topic、MessageQueue等路由信息(routing info)
  
• Product、Consumer客户端设置NameServer集群地址启动后，定时使命获取broker信息（broker上下线等操纵也会即时更新）
  
• Product根据负载均衡算法、Topic、MessageQueue和Broker信息找到要通信的Broker发送消息
  
• Broker收到Product消息后进行持久化
  
• Consumer根据再均衡算法得到自己要消耗的队列，再通过Broker信息通信获取消息进行消耗
  

在这个流程中，小菜有个疑问：为什么Product、Consumer获取Broker数据要通过NameServer通信？

• NameServer在架构中的作用如注册中央，管理服务注册与发现
  
• 架构解耦，将心跳/交互数据/判断状态等功能交给NameServer（注册中央）去做
  
• 在broker集群下，如果没有NameServer这种broker间必要互相进行心跳和同步汇总的数据，当节点繁多时增加带宽压力，另外broker宕机时还要增加机制来进行判断是否下线，并且product与consumer必要设置broker信息会非常多（必要网络通信）
  
• NameServer集群间节点无状态互不通信，提供高可用集群
  

Spring Boot 快速上手

RocketMQ的broker作为服务端，NameServer作为注册中央，与编写代码的接触比较少，较多的照旧生产者与消耗者（客户端）
经过大量的理论知识，我们知道MQ的大致流程，接下来使用SpringBoot编写代码实现Product和Consumer客户端
原生RocketMQ提供的生产者与消耗者API繁多并且使用时必要try catch、使用起来贫苦，企业级开辟通常会在其基础上进行封装常用的API
Spring Boot 框架作为脚手架，整合RocketMQ会非常快，并且还提供对应的RocketMQTemplate对原生API进行封装简化开辟

• 导入maven依靠
  

1. <dependency>
2.     <groupId>org.apache.rocketmq</groupId>
3.     <artifactId>rocketmq-spring-boot-starter</artifactId>
4.     <version>2.2.3</version>
5. </dependency>

复制代码

• 封装原生Product API
  

企业级开辟经常会对原生API进行封装，而其中的ServerProduct是自界说的类，组合原生默认的生产者DefaultMQProducer来封装API简化开辟
在这个过程中，通常会用设置文件的方式设置有关生产者的参数如：组名、nameserver地址、发送消息失败重试次数、发送消息超时时间等…
设置完参数后，启动生产者 producer.start()

1. public class ServerProduct {
3.     private DefaultMQProducer producer;
5.     public ServerProduct(String producerGroup) {
6.         producer = new DefaultMQProducer(producerGroup);
7.         init();
8.     }
10.     public ServerProduct() {
11.         producer = new DefaultMQProducer("Default_Server_Producer_Group");
12.         init();
13.     }
15.     private void init() {
16.         //初始化 主要通过配置文件的值进行set 最后启动生产者
17.         producer.setNamesrvAddr("127.0.0.1:9876");
18.         //...
20.         try {
21.             producer.start();
22.         } catch (MQClientException e) {
23.             throw new RuntimeException(e);
24.         }
25.     }
26. }

复制代码

启动生产者重要会去启动定时使命同步nameserver、broker数据，并初始化一些组件，后续用于客户端网络通信、负载均衡等
这些原理放到后文源码分析再具体聊聊~
然后再封装一个发送消息的API：
sendSyncMsg 发送同步消息API中第一个参数为topic（一级分类），第二个传输为tag（二级分类），第三个传输为消息体
内部会通过参数构建Message并使用原生API发送给Broker

1. public SendResult sendSyncMsg(String topic, String tag, String jsonBody) {
2.     Message message = new Message(topic, tag, jsonBody.getBytes(StandardCharsets.UTF_8));
3.     SendResult sendResult;
4.     try {
5.         sendResult = producer.send(message);
6.     } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
7.         throw new RuntimeException(e);
8.     }
9.     return sendResult;
10. }

复制代码

• 编写controller类，使用生产者调用API，发送到broker
  

1. @RequestMapping("/warn")
2. @RestController
3. @Slf4j
4. public class WarnController {
6.     private static final String topic = "TopicTest";
8.     @Autowired
9.     private ServerProduct producer;
11.     @GetMapping("/syncSend")
12.     public SendResult syncSend() {
13.         return producer.sendSyncMsg(topic, "tag", "sync hello world!");
14.     }
15. }

复制代码

• 消耗者订阅Topic
  

发送完消息后，消息会持久化到broker中，因此我们必要使用消耗者获取消息并进行消耗
企业级开辟时通常会使用注解的方式标识consumer必要订阅的信息，再通过分析注解的方式将数据注入的消耗者中，我们这里直接使用spring提供的注解@RocketMQMessageListener

1. @Component
2. @RocketMQMessageListener(topic = "TopicTest", consumerGroup = "warn_consumer_group")
3. public class WarnConsumer implements RocketMQListener<String> {
5.     @Override
6.     public void onMessage(String message) {
7.         // 处理消息
8.         System.out.println("Received message: " + message);
9.     }
10. }

复制代码

• 手动创建Topic：TopicTest 不要自动创建Topic 项目太大可能导致遗忘
  

使用Dashboard手动创建Topic

• 启动NameServer、Broker再启动SpringBoot服务，访问 /warn/syncSend 测试发送消息、消耗
  

NameServer、Broker的摆设可以查看官方文档
至此我们履历过消息的生产与消耗，但在消息的“一生”中还可能出现各种各样的环境：

• 发送消息的方式：平凡（同步、异步、单向）/顺序/延迟/批量/事务消息等…
  
• 消耗消息的方式：push/pull消耗、集群/广播模式…
  
• 如何保证消息不丢失？
  
• 消息是如何高效持久化的？
  
• 如何保证消耗幂等？如何解决消息堆积、延时？
  

这些环境后文都会由浅入深一一解决，查看RocketMQ实现原理，并从原理中体验出计划的思想，再去查看其他的消息中央件~
总结

消息中央件通常有削峰填谷、异步通信、架构解耦、高性能、高可用、集群扩展、负载均衡等相干特点，同时项目中引入消息中央件也会增加调用链路、系统复杂度
RocketMQ由NameServer、Broker、Product、Consumer等集群组成
其中Broker作为服务端，负责接收消息、对消息进行高效持久化、消耗消息时高效查询
界说Topic对消息进行分类，为了提升水平扩展写入本领，Topic下可以设置MessageQueue队列，消息作为数据载体存储在队列中，等待被消耗
为了满足高可用，相同的Topic会被放到不同的master broker，避免”所有坤蛋都在同一个篮子“中
NameServer集群作为“注册中央”，节点无状态之间互不通信，只与Broker集群心跳同时更新路由信息，比及Product、Consumer定时通信时再将Broker信息进行传输
Product为消息生产方，通过与NameServer获取的Broker中Topic、队列ID等信息，使用负载均衡算法后找到对应Broker进行通信
Consumer为消息消耗者，根据再均衡负载均衡得到自己负责消耗的队列，再通过Broker获取消息进行消耗
最后（点赞、收藏、关注求求啦~）

本篇文章被收入专栏 消息中央件，感爱好的同学可以连续关注喔
本篇文章笔记以及案例被收入 Gitee-CaiCaiJava、 Github-CaiCaiJava，除此之外另有更多Java进阶相干知识，感爱好的同学可以starred连续关注喔~
有什么问题可以在评论区交流，如果觉得菜菜写的不错，可以点赞、关注、收藏支持一下~
关注菜菜，分享更多技术干货，公众号：菜菜的后端私房菜

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。