读完 RocketMQ 源码,我学会了如何优雅的创建线程

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 读完 RocketMQ 源码,我学会了如何优雅的创建线程


RocketMQ 是一款开源的分布式消息系统,基于高可用分布式集群技术,提供低延时、高可靠的消息发布与订阅服务。

这篇文章,笔者整理了 RocketMQ 源码中创建线程的几点技巧,希望大家读完之后,能够有所收获。

1. 创建单线程

首先我们先温习下常用的创建单线程的两种方式:

  • 实现 Runnable 接口
  • 继承 Thread 类

▍一、实现 Runnable 接口

图中,MyRunnable 类实现了 Runnable 接口的 run 方法,run 方法中定义具体的任务代码或处理逻辑,而Runnable 对象是作为线程构造函数的参数。

▍二、 继承 Thread 类

线程实现类直接继承 Thread ,本质上也是实现 Runnable 接口的 run 方法。

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

2. 单线程抽象类

创建单线程的两种方式都很简单,但每次创建线程代码显得有点冗余,于是 RocketMQ 里实现了一个抽象类 ServiceThread 。

抽象类 ServiceThread

我们可以看到抽象类中包含了如下核心方法:

  1. 定义线程名;
  2. 启动线程;
  3. 关闭线程。

下图展示了 RocketMQ 众多的单线程实现类。

实现类的编程模版类似 :

我们仅仅需要继承抽象类,并实现 getServiceNamerun 方法即可。启动的时候,调用 start 方法 , 关闭的时候调用 shutdown 方法。

基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

3. 线程池原理

线程池是一种基于池化思想管理线程的工具,线程池维护着多个线程,等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用,还能防止过分调度。关注公z号:码猿技术专栏,回复关键词:1111 获取阿里内部性能调优手册

JDK中提供的 ThreadPoolExecutor 类,是我们最常使用的线程池类。

ThreadPoolExecutor构造函数

参数名 作用
corePoolSize 队列没满时,线程最大并发数
maximumPoolSizes 队列满后线程能够达到的最大并发数
keepAliveTime 空闲线程过多久被回收的时间限制
unit keepAliveTime 的时间单位
workQueue 阻塞的队列类型
threadPoolFactory 改变线程的名称、线程组、优先级、守护进程状态
RejectedExecutionHandler 超出 maximumPoolSizes + workQueue 时,任务会交给RejectedExecutionHandler来处理

任务的调度通过执行 execute方法完成,方法的核心流程如下:

  1. 如果 workerCount < corePoolSize,创建并启动一个线程来执行新提交的任务。
  2. 如果 workerCount >= corePoolSize,且线程池内的阻塞队列未满,则将任务添加到该阻塞队列中。
  3. 如果 workerCount >= corePoolSize && workerCount < maximumPoolSize,且线程池内的阻塞队列已满,则创建并启动一个线程来执行新提交的任务。
  4. 如果 workerCount >= maximumPoolSize,并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。

4. 线程池封装

在 RocketMQ 里 ,网络请求都会携带命令编码,每种命令映射对应的处理器,而处理器又会注册对应的线程池。

当服务端 Broker 接收到发送消息命令时,都会有单独的线程池 sendMessageExecutor 来处理这种命令请求。

基于 ThreadPoolExecutor 做了一个简单的封装 ,BrokerFixedThreadPoolExecutor 构造函数包含六个核心参数:

  1. 核心线程数和最大线程数相同 ,数量是:cpu核数和4比较后的最小值;
  2. 空闲线程的回收的时间限制,默认1分钟;
  3. 发送消息队列,有界队列,默认10000;
  4. 线程工厂 ThreadFactoryImpl ,定义了线程名前缀:SendMessageThread_ 。

RocketMQ 实现了一个简单的线程工厂:ThreadFactoryImpl ,线程工厂可以定义线程名称,以及是否是守护线程 。

线程工厂

开源项目 Cobar ,Xmemcached,Metamorphosis 中都有类似线程工厂的实现 。

5. 线程名很重要

线程名很重要,线程名很重要,线程名很重要 ,重要的事情说三遍。

我们看到 RocketMQ 中,无论是单线程抽象类还是多线程的封装都会配置线程名 ,因为通过线程名,非常容易定位问题,从而大大提升解决问题的效率。

定位的媒介常见有两种:日志文件堆栈记录

▍一、日志文件

经常处理业务问题的同学,一定都经常与日志打交道。

  • 查看 ERROR 日志,追溯到执行线程, 要是线程池隔离做的好,基本可以判断出哪种业务场景出了问题;
  • 通过查看线程打印的日志,推断线程调度是否正常,比如有的定时任务线程打印了开始,没有打印结束,推论当前线程可能已经挂掉或者阻塞。

▍二、堆栈记录

jstack 是 java 虚拟机自带的一种堆栈跟踪工具 ,主要用来查看 Java 线程的调用堆栈,线程快照包含当前 java 虚拟机内每一条线程正在执行的方法堆栈的集合,可以用来分析线程问题。

jstack -l 进程pid

笔者查看线程堆栈,一般关注如下几点:

  1. 当前 jvm 进程中的线程数量和线程分类是否在预期的范围内;
  2. 系统接口超时或者定时任务停止的异常场景下 ,分析堆栈中是否有锁未释放,或者线程一直等待网络通讯响应;
  3. 分析 jvm 进程中哪个线程占用的 CPU 最高。

6. 总结

本文是RocketMQ 系列文章的开篇,和朋友们简单聊聊 RocketMQ 源码里创建线程的技巧。

  1. 单线程抽象类 ServiceThread
    使用者只需要实现业务逻辑以及定义线程名即可 ,不需要写冗余的代码。
  2. 线程池封装
    适当封装,定义线程工厂,并合理配置线程池参数。
  3. 线程名很重要
    文件日志,堆栈记录配合线程名能大大提升解决问题的效率。

RocketMQ 的多线程编程技巧很多,比如线程通讯,并发控制,线程模型等等,后续的文章会一一为大家展现。



相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
13天前
|
缓存 监控 Java
Java线程池提交任务流程底层源码与源码解析
【11月更文挑战第30天】嘿,各位技术爱好者们,今天咱们来聊聊Java线程池提交任务的底层源码与源码解析。作为一个资深的Java开发者,我相信你一定对线程池并不陌生。线程池作为并发编程中的一大利器,其重要性不言而喻。今天,我将以对话的方式,带你一步步深入线程池的奥秘,从概述到功能点,再到背景和业务点,最后到底层原理和示例,让你对线程池有一个全新的认识。
42 12
|
21天前
|
消息中间件 存储 监控
深度写作:深入源码理解MQ长轮询优化机制
【11月更文挑战第22天】在分布式系统中,消息队列(Message Queue, MQ)扮演着至关重要的角色。MQ不仅实现了应用间的解耦,还提供了异步消息处理、流量削峰等功能。而在MQ的众多特性中,长轮询(Long Polling)机制因其能有效提升消息处理的实时性和效率,备受关注。
55 12
|
21天前
|
消息中间件 存储 Java
深入源码理解MQ长轮询优化机制
【11月更文挑战第22天】在分布式系统中,消息队列(MQ)作为一种重要的中间件,广泛应用于解耦、异步处理、流量削峰等场景。其中,延时消息和定时消息作为MQ的高级功能,能够进一步满足复杂的业务需求。为了实现这些功能,MQ系统需要进行一系列优化,长轮询机制便是其中的关键一环。本文将深入探讨MQ如何设计延时消息和定时消息的优化机制,特别是长轮询机制的实现原理及其在Java中的模拟实现。
32 2
|
3月前
|
存储 缓存 Java
什么是线程池?从底层源码入手,深度解析线程池的工作原理
本文从底层源码入手,深度解析ThreadPoolExecutor底层源码,包括其核心字段、内部类和重要方法,另外对Executors工具类下的四种自带线程池源码进行解释。 阅读本文后,可以对线程池的工作原理、七大参数、生命周期、拒绝策略等内容拥有更深入的认识。
147 29
什么是线程池?从底层源码入手,深度解析线程池的工作原理
|
2月前
|
存储 运维 API
源码解密协程队列和线程队列的实现原理(一)
源码解密协程队列和线程队列的实现原理(一)
37 1
|
2月前
|
存储 安全 API
源码解密协程队列和线程队列的实现原理(二)
源码解密协程队列和线程队列的实现原理(二)
35 1
|
6月前
|
消息中间件 数据可视化 Go
Rabbitmq 搭建使用案例 [附源码]
Rabbitmq 搭建使用案例 [附源码]
49 0
|
2月前
|
传感器 数据可视化 网络协议
DIY可视化整合MQTT生成UniApp源码
DIY可视化整合MQTT生成UniApp源码
47 0
|
4月前
|
算法 安全 Java
深入解析Java多线程:源码级别的分析与实践
深入解析Java多线程:源码级别的分析与实践
|
6月前
|
缓存 并行计算 Java
重温JAVA线程池精髓:Executor、ExecutorService及Executors的源码剖析与应用指南
重温JAVA线程池精髓:Executor、ExecutorService及Executors的源码剖析与应用指南