RocketMQ中的线程池是如何创建的？

「爱情、让人受尽委屈。」 2023-09-23 18:51 49阅读 0赞

# 前言 #

大家好，今天主要来和大家聊一聊RocketMQ中的线程池是如何创建的，如何设置线程池数量，同时也可以从中去学习到一些线程池的实践和需要注意的一些细节。

## RocketMQ在哪些地方使用到了线程池？ ##

在RocketMQ中存在了大量的对线程池的使用，从消息的生产到投递Broker中，到最后的消息消费每一个环节中都大量使用到线程池的地方，下面我们拿出几个不同类型的线程池来看一看。

在 NameServer的路由注册和剔除中，多次使用到了定时线程池

### 定时线程池 ###

private final ScheduledExecutorService scheduledExecutorService =
    	Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
    		"NSScheduledThread"));
    复制代码

// 定时任务 每10s扫描一次Broker，移除失活Broker
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    	
    	@Override
    	public void run() {
    		NamesrvController.this.routeInfoManager.scanNotActiveBroker();
    	}
    }, 5, 10, TimeUnit.SECONDS);
    
    //定时任务，每隔30s向集群中所有NameServer发送心跳包
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    	
    	@Override
    	public void run() {
    		try {
    			BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
    		} catch (Throwable e) {
    			log.error("registerBrokerAll Exception", e);
    		}
    	}
    }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
    复制代码

### 线程池newFixedThreadPool ###

FixedThreadPool常用于创建一个固定大小的线程池，

它的特点就是核心线程数量与最大线程数量一致，采用无界的阻塞队列 LinkedBlockingQueue，并且没有设置队列的大小默认是Integer.MAX\_VALUE，适用于负载较重的场景

private ExecutorService remotingExecutor;
    
    this.remotingExecutor =
                Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
    
    // 用来设置接收到消息后的处理方法
    this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
    复制代码

### 消息发送初始化默认异步发送者线程池 ###

核心线程数与最大线程数设置均为 Runtime.getRuntime().availableProcessors() ，可用的计算资源

阻塞队列设置为一个初始化50000长度的阻塞队列

keepAliveTime设置60s，超过则时间空闲的线程将被终止

private final ExecutorService defaultAsyncSenderExecutor;
    
    private final BlockingQueue<Runnable> asyncSenderThreadPoolQueue;
    
    this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
    
    this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
    	Runtime.getRuntime().availableProcessors(),
    	Runtime.getRuntime().availableProcessors(),
    	1000 * 60,
    	TimeUnit.MILLISECONDS,
    	this.asyncSenderThreadPoolQueue,
    	new ThreadFactory() {
    		private AtomicInteger threadIndex = new AtomicInteger(0);
    
    		@Override
    		public Thread newThread(Runnable r) {
    			return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet());
    		}
    	});
    复制代码

### 消费端拉取消息线程池 ###

我们重点来看一下消费端的线程池是如何创建，它可以说是整个RocketMQ中最关键的一个线程池

为了提高消费速度，我们通常有两种方式来提高消费并行度

1.  同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度
2.  提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。

如何创建？

在消息监听的时候，利用线程池进行不断的拉取消息

![format_png][]

提交消费请求，消息提交到内部的线程池

DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
        pullResult.getMsgFoundList(),
        processQueue,
        pullRequest.getMessageQueue(),
        dispatchToConsume);
    复制代码

参数设置

创建内部线程池，核心参数核心线程数和最大线程数，主要是根据配置来进行设置

设置线程池名称以 ConsumeMessageThread\_ 开头的，利于排查问题

阻塞队列是一个无界的阻塞队列LinkedBlockingQueue

private final BlockingQueue<Runnable> consumeRequestQueue;
    
    this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();
    
    
    this.consumeExecutor = new ThreadPoolExecutor(
        this.defaultMQPushConsumer.getConsumeThreadMin(),
        this.defaultMQPushConsumer.getConsumeThreadMax(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        this.consumeRequestQueue,
        new ThreadFactoryImpl(consumeThreadPrefix));
    复制代码

通过RocketMQ的源码，我们看到 consumeExecutor 线程池的创建也是非常简单的

## 如果想要修改线程池参数，需要注意什么？ ##

根据线程池的原理我们知道，只有阻塞队列为满的情况下，不会创建临时线程

![format_png 1][]

所以线程池内部持有的队列为一个无界队列，导致 consumeThreadMax 大于 consumeThreadMin，线程个数最大也只能 consumeThreadMin 个线程数量

什么时候需要修改？

在正常的业务场景中，启动应用之后，我们就不会再修改消费者线程数，但有可能突发业务高峰导致消息堆积，这时候我们就需要调整单个 Consumer 的消费并行线程数。

如何修改线程数？

1.  修改线程池后，重新启动消费者，缺点是参数不易评估，随着业务的并发提升，需要频繁的重启服务来更改线程数，这势必会带来一定的造成影响。
2.  官方也为我们提供了修改线程数的方法，当更新的线程数大于0且小于 Short.MAX\_VALUE 且小于最大线程数，则更新核心线程数。

JDK允许线程池使用方通过ThreadPoolExecutor的实例来动态设置线程池的核心策略

@Override
    public void updateCorePoolSize(int corePoolSize) {
        if (corePoolSize > 0
            && corePoolSize <= Short.MAX_VALUE
            && corePoolSize < this.defaultMQPushConsumer.getConsumeThreadMax()) {
            this.consumeExecutor.setCorePoolSize(corePoolSize);
        }
    }
    复制代码

这两种方式都存在一定的痛点

1.  线程数量随着业务的变动，需要修改代码
2.  在springBoot和SpringCloud Stream下，对线程池参数变更不是很友好
3.  不能通过管理界面，直接动态修改线程池参数

针对上面的痛点问题，我们可以考虑封装线程池动态参数调整，首先肯定原来代码是毫无侵入性的，

同时通过管理页面对不同消费者组的线程池进行管理自由的随着业务波动进行平滑修改，降低线程池参数修改的成本。

[format_png]: https://img-blog.csdnimg.cn/img_convert/b95431efac2c1f7159fd95b484a94c7b.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/15f0e9ec006da5df0b6a134b8a14321b.webp?x-oss-process=image/format,png