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

时间：2023-01-02

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

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

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

　　实现 Runnable 接口继承 Thread 类

　　▍一、实现 Runnable 接口

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

　　▍二、继承 Thread 类

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

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

　　抽象类 ServiceThread

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

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

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

　　实现类的编程模版类似：

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

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

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

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

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

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