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

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

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

1 创建单线程

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

▍一、实现 Runnable 接口

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

▍二、继承 Thread 类

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

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

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

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

实现类的编程模版类似：

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

线程池是一种基于池化思想管理线程的工具，线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。

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

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

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

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