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

红太狼 2023-10-01 23:59 4阅读 0赞

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

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

![30b79f9d98811eed005c8018ffa6ccdb.png][]

## 1 创建单线程 ##

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

*  实现 Runnable 接口
 *  继承 Thread 类

**▍一、实现 Runnable 接口**

![bbbecd6d6345de1f52858377c318f1ab.png][]

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

**▍二、 继承 Thread 类**

![2659c2fb6e892f09f4db0fdc616e5e8f.png][]

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

## 2 单线程抽象类 ##

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

![a878630c276a0e1ccc176b3cf9a362f6.png][]

抽象类 ServiceThread

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

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

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

![e8d43909905a170eaaeeafa718310416.jpeg][]

实现类的编程模版类似 ：

![a17cf646e124727dc1229006bf0915ee.png][]

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

## 3 线程池原理 ##

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

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

![9b2ddb9f4601a811d6592f2864068a87.png][]

ThreadPoolExecutor构造函数

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

![7d5bfc694290dbdcaa9a52ba7ef43043.png][]

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

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

![9a8bd6ac33a0e793068dabc871828c0a.png][]

## 4 线程池封装 ##

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

![7edf07aba743151f5b4b11c3b7dae2bf.png][]

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

![02b7fe50ba54983f4a61ac4508c28af3.png][]

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

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

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

![ed585aed98d60d95b7c1ebd619e75a44.png][]

线程工厂

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

## 5 线程名很重要 ##

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

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

定位的媒介常见有两种：**日志文件**和**堆栈记录**。

**▍一、日志文件**

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

![67cc987a4a0b47004eecc18401d63b8e.png][]

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

**▍二、堆栈记录**

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

jstack -l 进程pid

![27864afe2038d7fac016fa2359f71d89.png][]

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

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

## 6 总结 ##

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

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

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

[30b79f9d98811eed005c8018ffa6ccdb.png]: https://img-blog.csdnimg.cn/img_convert/30b79f9d98811eed005c8018ffa6ccdb.png
[bbbecd6d6345de1f52858377c318f1ab.png]: https://img-blog.csdnimg.cn/img_convert/bbbecd6d6345de1f52858377c318f1ab.png
[2659c2fb6e892f09f4db0fdc616e5e8f.png]: https://img-blog.csdnimg.cn/img_convert/2659c2fb6e892f09f4db0fdc616e5e8f.png
[a878630c276a0e1ccc176b3cf9a362f6.png]: https://img-blog.csdnimg.cn/img_convert/a878630c276a0e1ccc176b3cf9a362f6.png
[e8d43909905a170eaaeeafa718310416.jpeg]: https://img-blog.csdnimg.cn/img_convert/e8d43909905a170eaaeeafa718310416.jpeg
[a17cf646e124727dc1229006bf0915ee.png]: https://img-blog.csdnimg.cn/img_convert/a17cf646e124727dc1229006bf0915ee.png
[9b2ddb9f4601a811d6592f2864068a87.png]: https://img-blog.csdnimg.cn/img_convert/9b2ddb9f4601a811d6592f2864068a87.png
[7d5bfc694290dbdcaa9a52ba7ef43043.png]: https://img-blog.csdnimg.cn/img_convert/7d5bfc694290dbdcaa9a52ba7ef43043.png
[9a8bd6ac33a0e793068dabc871828c0a.png]: https://img-blog.csdnimg.cn/img_convert/9a8bd6ac33a0e793068dabc871828c0a.png
[7edf07aba743151f5b4b11c3b7dae2bf.png]: https://img-blog.csdnimg.cn/img_convert/7edf07aba743151f5b4b11c3b7dae2bf.png
[02b7fe50ba54983f4a61ac4508c28af3.png]: https://img-blog.csdnimg.cn/img_convert/02b7fe50ba54983f4a61ac4508c28af3.png
[ed585aed98d60d95b7c1ebd619e75a44.png]: https://img-blog.csdnimg.cn/img_convert/ed585aed98d60d95b7c1ebd619e75a44.png
[67cc987a4a0b47004eecc18401d63b8e.png]: https://img-blog.csdnimg.cn/img_convert/67cc987a4a0b47004eecc18401d63b8e.png
[27864afe2038d7fac016fa2359f71d89.png]: https://img-blog.csdnimg.cn/img_convert/27864afe2038d7fac016fa2359f71d89.png