【Alibaba中间件技术系列】「RocketMQ技术专题」带你一起去探索RocketMQ服务架构的线程模型分析

简介: 【Alibaba中间件技术系列】「RocketMQ技术专题」带你一起去探索RocketMQ服务架构的线程模型分析

前言介绍


RocketMQ 是个消息服务器,也是个网络服务器。接下来我们将从网络 IO 模型,线程模型,看看 RocketMQ 是如何设计的。




IO 模型


RocketMQ 使用了 Netty 作为网络通信框架,自然而然使用了 Reactor 模型,或者说 Select 模型、Epoll 模型。即一个线程管理 N 个 Socket 的模式,此模式可管理海量连接,基本是所有网络服务器的首选。



配置相关的RocketMQ的配置


RocketMQ的Boss线程数为 1, Worker 线程数为 CPU * 2.

在说线程模型之前,先看看 RocketMQ 如何设计 Server 接口的。

image.png

RemotingService作为顶层接口,定义了 启动和关闭,另外还有注册 RPC 钩子,职责简单。他的两个子接口 RemotingServer 和 RemotingClient 各自增加了自己的抽象接口。


  • Server 专属的 localListenPort 和 getProcessorPair
  • Client 专属的 getNameServerAddressList 等


注意:两者都有 invokeSync 方法,但,参数不同,这也是因为他们自身的角色不同所影响的。


至于NettyRemotingAbstract 抽象类,这只是个简单的”抽取重复代码”的“简单操作”。

再下面,就是具体实现类,每个类,都有内部类,都是Netty各种Handler 的实现:

image.png

  • NettyConnectManageHandler 负责处理 注册,连接,异常等事件,继承自 ChannelDuplexHandler。
  • NettyServerHandler 则是关键的业务处理类,处理真正的 Msg,继承自 SimpleChannelInboundHandler。
  • HandshakeHandler 负责处理握手程序,这里就不解释了。


以上 3 个是 Server 端的 Handler,都是NettyRemotingServer的内部类。



线程模型


NettyServerHandler作为处理业务的关键类,每个 worker 线程都有自己的单独实例,但该类只是做个包装或者桥接而已,作用不大, NettyRemotingServer才是关键。


当 Request 进入到 Server 中,MQ 会根据 请求类型 code 找到对应的处理器,MQ 有多种处理器,如下:

image.png


他们都继承自 NettyRequestProcessor 接口:

image.png

此接口只有 2 个方法,处理请求和拒绝请求,处理请求的参数是 Netty 的 context 和自身的RemotingCommand 对象,这是个大对象:

image.png


RemotingCommand 的成员变量,这里说下 flag 的作用,其他就不说了。


flag表示这次请求是什么类型。


  • 倒数第一位,0 表示请求,1 表示返回。
  • 倒数第二位,1 表示 oneWay。



NettyRequestProcessor


刚刚提到 NettyRequestProcessor ,这是个处理器,在 MQ 中,每个 NettyRequestProcessor 都绑定了一个线程池,在 MQ 的抽象里,有个 Pair 对象,如下:

public class Pair<T1, T2> {
    private T1 object1;
    private T2 object2;
    public Pair(T1 object1, T2 object2) { this.object1 = object1; 
                                         this.object2 = object2;}
    public T1 getObject1() { return object1;}
    public void setObject1(T1 object1) {this.object1 = object1; }
    public T2 getObject2() { return object2;}
    public void setObject2(T2 object2) {  this.object2 = object2;}
}
复制代码

同时,还有个 Hash 表,用 code 映射了 Pair。如此,就实现了:通过请求 code 找到“线程池和处理这种请求的处理器”,然后,提交一个任务到该线程池,任务中,会调用该处理器的 processRequest 方法,或 rejectRequest 方法。

image.png

上图中,为处理请求的关键步骤。执行钩子就不说了,我们知道,设计代码时,关键步骤都加钩子,便于扩展和以后加代码。


其中,会调用 processRequest 方法,执行具体业务,并得到返回值。然后使用 netty 的 ctx 对象,将返回值直接写回 Socket。


如果发生错误了,也将错误构造成消息,写回客户端。


注意,这里一直有个操作 就是 response.setOpaque(opaque) ,就是设置请求 ID,这是 IO 多路复用的关键。


Netty 每次请求,都会调用 NettyRemotingServer 的 processRequestCommand 方法。


而 NettyRemotingServer 保存了请求 code 和 Pair<处理器,线程池> 的hash 映射表。


每次请求,根据 code 找到线程池,生成一个新任务,提交到线程池,任务里,会执行“处理器” 的processRequest 方法得到返回值,最后写回客户端。


MQ 为每种类型的任务,使用了不同的线程池,即线程池隔离。同时,也根据每种不同的任务类型,设置了不同的线程池参数。




参数介绍


  • Send 发送消息任务,线程池大小是1。
  • pull 拉取消息任务,线程池大小是 16 + CPU*2
  • query 查询任务,线程池大小是 8 + CPU*2;


当然还有其他的,这里就不枚举了,注意:大部分线程池都是多线程,只有 send 任务默认是单线程。


send 操作是个写操作,最后是要上锁的,虽然锁的粒度已经足够小,但仍然是有锁的。如果是有锁的,多线程的是不划算的。这也是 RocketMQ 的设计决定只写一个 CommitLog。


能像Kafka一样,同时写多个文件,是不是就可以利用多线程了呢?


当然,这里不是说多线程一定好,只是表达另外一种思路。如果单线程就能触发 MQ 瓶颈,多线程也没啥意义。


总结


image.png


相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
4月前
|
存储 缓存 安全
某鱼电商接口架构深度剖析:从稳定性到高性能的技术密码
某鱼电商接口架构揭秘:分层解耦、安全加固、性能优化三维设计,实现200ms内响应、故障率低于0.1%。详解三层架构、多引擎存储、异步发布、WebSocket通信与全链路防护,助力开发者突破电商接口“三难”困境。
|
4月前
|
Java Linux 虚拟化
【Docker】(1)Docker的概述与架构,手把手带你安装Docker,云原生路上不可缺少的一门技术!
1. Docker简介 1.1 Docker是什么 为什么docker会出现? 假定您在开发一款平台项目,您的开发环境具有特定的配置。其他开发人员身处的环境配置也各有不同。 您正在开发的应用依赖于您当前的配置且还要依赖于某些配置文件。 您的企业还拥有标准化的测试和生产环境,且具有自身的配置和一系列支持文件。 **要求:**希望尽可能多在本地模拟这些环境而不产生重新创建服务器环境的开销 问题: 要如何确保应用能够在这些环境中运行和通过质量检测? 在部署过程中不出现令人头疼的版本、配置问题 无需重新编写代码和进行故障修复
429 2
|
4月前
|
设计模式 缓存 安全
【JUC】(6)带你了解共享模型之 享元和不可变 模型并初步带你了解并发工具 线程池Pool,文章内还有饥饿问题、设计模式之工作线程的解决于实现
JUC专栏第六篇,本文带你了解两个共享模型:享元和不可变 模型,并初步带你了解并发工具 线程池Pool,文章中还有解决饥饿问题、设计模式之工作线程的实现
260 2
|
4月前
|
存储 人工智能 搜索推荐
拔俗AI助教系统:基于大模型与智能体架构的新一代教育技术引擎
AI助教融合大语言模型、教育知识图谱、多模态感知与智能体技术,重构“教、学、评、辅”全链路。通过微调LLM、精准诊断错因、多模态交互与自主任务规划,实现个性化教学。轻量化部署与隐私保护设计保障落地安全,未来将向情感感知与教育深度协同演进。(238字)
454 0
|
7月前
|
消息中间件 数据管理 Serverless
阿里云消息队列 Apache RocketMQ 创新论文入选顶会 ACM FSE 2025
阿里云消息团队基于 Apache RocketMQ 构建 Serverless 消息系统,适配多种主流消息协议(如 RabbitMQ、MQTT 和 Kafka),成功解决了传统中间件在可伸缩性、成本及元数据管理等方面的难题,并据此实现 ApsaraMQ 全系列产品 Serverless 化,助力企业提效降本。
|
5月前
|
消息中间件 Java Kafka
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
本文深入解析了 Kafka 和 RabbitMQ 两大主流消息队列在 Spring 微服务中的应用与对比。内容涵盖消息队列的基本原理、Kafka 与 RabbitMQ 的核心概念、各自优势及典型用例,并结合 Spring 生态的集成方式,帮助开发者根据实际需求选择合适的消息中间件,提升系统解耦、可扩展性与可靠性。
355 1
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
944 87
|
消息中间件 安全 Java
云消息队列RabbitMQ实践解决方案评测
一文带你详细了解云消息队列RabbitMQ实践的解决方案优与劣
436 88
|
消息中间件 存储 Kafka
MQ 消息队列核心原理,12 条最全面总结!
本文总结了消息队列的12个核心原理,涵盖消息顺序性、ACK机制、持久化及高可用性等内容。关注【mikechen的互联网架构】,10年+BAT架构经验倾囊相授。

相关产品

  • 云消息队列 MQ