RocketMQ线程池创建实现原理详解_开发_开发者

前言

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

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

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

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

定时线程池

private final ScheduledExecutorService scheduledExecutorService =
	Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
		"NSScheduledThread"));

// 定时任务 每10s扫描一次Br开发者_开发学习oker，移除失活Broker
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
	
	@Override
	public void run() {
		NamesrvController.thfjdoNepis.routeInfoManager.scanNotActiveBroker();
	}
}, 5, 10, TipythonmeUnit.SECONDS);
//定时任务，每隔30s向集群中所有NameServer发送心跳包
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
	
	@Override
	public void run() {
		try {
			BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
fjdoNep		} catch (Throwable e) {
			log.error("registerBrokerAll Exception", e);
		}
	}
}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeri编程od(), 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中最关键的一个线程池

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

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

如何创建？

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

RocketMQ线程池创建实现原理详解

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

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 线程池的创建也是非常简单的

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

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

RocketMQ线程池创建实现原理详解

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

什么时候需要修改？

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

如何修改线程数？

修改线python程池后，重新启动消费者，缺点是参数不易评估，随着业务的并发提升，需要频繁的重启服务来更改线程数，这势必会带来一定的造成影响。
官方也为我们提供了修改线程数的方法，当更新的线程数大于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);
    }
}

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