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

点击上方“芋道源码”，选择“设为星标”

管她前浪，还是后浪？

能浪的浪，才是好浪！

每天 10:33 更新文章，每天掉亿点点头发...

源码精品专栏

来源：勇哥java实战分享

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

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

1. 创建单线程

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

实现 Runnable 接口
继承 Thread 类

▍一、实现 Runnable 接口

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

▍二、继承 Thread 类

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

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址：https://github.com/YunaiV/ruoyi-vue-pro
视频教程：https://doc.iocoder.cn/video/

2. 单线程抽象类

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

抽象类 ServiceThread

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

定义线程名；
启动线程；
关闭线程。

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

实现类的编程模版类似：

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

基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址：https://github.com/YunaiV/yudao-cloud
视频教程：https://doc.iocoder.cn/video/

3. 线程池原理

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

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

ThreadPoolExecutor构造函数

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

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

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