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依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.3</version>
</dependency>
- 封装原生Product API
企业级开发常常会对原生API进行封装，而其中的ServerProduct是自定义的类，组合原生默认的生产者DefaultMQProducer来封装API简化开发
在这个过程中，通常会用配置文件的方式配置有关生产者的参数如：组名、nameserver地址、发送消息失败重试次数、发送消息超时时间等…
设置完参数后，启动生产者 producer.start()
public class ServerProduct {
private DefaultMQProducer producer;
public ServerProduct(String producerGroup) {
producer = new DefaultMQProducer(producerGroup);
init();
}
public ServerProduct() {
producer = new DefaultMQProducer("Default_Server_Producer_Group");
init();
}
private void init() {
//初始化 主要通过配置文件的值进行set 最后启动生产者
producer.setNamesrvAddr("127.0.0.1:9876");
//...
try {
producer.start();
} catch (MQClientException e) {
throw new RuntimeException(e);
}
}
}
启动生产者主要会去启动定时任务同步nameserver、broker数据，并初始化一些组件，后续用于客户端网络通信、负载均衡等
这些原理放到后文源码解析再具体聊聊~
然后再封装一个发送消息的API：
sendSyncMsg 发送同步消息API中第一个参数为topic（一级分类），第二个传输为tag（二级分类），第三个传输为消息体
内部会通过参数构建Message并使用原生API发送给Broker
public SendResult sendSyncMsg(String topic, String tag, String jsonBody) {
Message message = new Message(topic, tag, jsonBody.getBytes(StandardCharsets.UTF_8));
SendResult sendResult;
try {
sendResult = producer.send(message);
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
throw new RuntimeException(e);
}
return sendResult;
}
- 编写controller类，使用生产者调用API，发送到broker
@RequestMapping("/warn")
@RestController
@Slf4j
public class WarnController {
private static final String topic = "TopicTest";
@Autowired
private ServerProduct producer;
@GetMapping("/syncSend")
public SendResult syncSend() {
return producer.sendSyncMsg(topic, "tag", "sync hello world!");
}
}
- 消费者订阅Topic
发送完消息后，消息会持久化到broker中，因此我们需要使用消费者获取消息并进行消费
企业级开发时通常会使用注解的方式标识consumer需要订阅的信息，再通过解析注解的方式将数据注入的消费者中，我们这里直接使用spring提供的注解@RocketMQMessageListener
@Component
@RocketMQMessageListener(topic = "TopicTest", consumerGroup = "warn_consumer_group")
public class WarnConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
// 处理消息
System.out.println("Received message: " + message);
}
}
- 手动创建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持续关注喔~
有什么问题可以在评论区交流，如果觉得菜菜写的不错，可以点赞、关注、收藏支持一下~
关注菜菜，分享更多技术干货，公众号：菜菜的后端私房菜
- 点赞
- 收藏
- 关注作者
评论（0）