您现在的位置是：亿华云 > 系统运维

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

亿华云2025-10-09 11:08:34【系统运维】2人已围观

简介RocketMQ 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时、高可靠的消息发布与订阅服务。这篇文章，笔者整理了 RocketMQ 源码中创建线程的几点技巧，希望大家读完之后，能够

RocketMQ 是读完一款开源的分布式消息系统，基于高可用分布式集群技术，源码雅提供低延时、学会线程高可靠的何优消息发布与订阅服务。

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

一、何优创建单线程

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

实现 Runnable 接口

继承 Thread 类

1.实现 Runnable 接口

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

2.继承 Thread 类

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

二、单线程抽象类

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

抽象类 ServiceThread

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

定义线程名；启动线程；关闭线程。站群服务器

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

实现类的编程模版类似：

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

三、线程池原理

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

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，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。