ElasticSearch 线程池类型分析之SizeBlockingQueue

ElasticSearch 线程池类型分析之SizeBlockingQueue

尽管前面写好几篇ES线程池分析的文章(见文末参考链接),但都不太满意。但从ES的线程池中了解到了不少JAVA线程池的使用技巧,于是忍不住再写一篇(ES6.3.2版本的源码)。文中给出的每个代码片断,都标明了这些代码是来自哪个类的哪个方法。

ElasticSearch里面一共有四种类型的线程池,源码:ThreadPool.ThreadPoolType

        DIRECT("direct"),
FIXED("fixed"),
FIXED_AUTO_QUEUE_SIZE("fixed_auto_queue_size"),
SCALING("scaling");

GET、SEARCH、WRITE、INDEX、FLUSH...等各种操作是交由这些线程池实现的。为什么定义不同类型的线程池呢?举个最简单的例子:程序里面有IO密集型任务,也有CPU密集型任务,这些任务都提交到一个线程池中执行?还是根据任务的执行特点将CPU密集型的任务都提交到一个线程池,IO密集型任务都提交到另一个线程池执行?

不同种类的操作(INDEX、SEARCH...)交由不同类型的线程池执行是有很多好处的:

  1. 互不影响:INDEX操作频繁时,并不会影响SEARCH操作的执行。
  2. 资源合理利用(提升性能):如果只有一个线程池来处理所有的操作,线程池队列长度配置为多大合适?线程的数量配置多少合适?这些操作难道都要共用一个拒绝策略吗?线程执行过程中出现异常了,针对不同类型的操作,异常处理方案也是不一样的,显然:只有一个线程池(或者说只有一种类型的线程池),是无法满足这些需求的。

再来说一下ES中的线程池都是如何创建的?

ES节点启动时,执行Node类的构造方法 :org.elasticsearch.node.Node.Node(org.elasticsearch.env.Environment, java.util.Collection<java.lang.Class<? extends org.elasticsearch.plugins.Plugin>>)

final ThreadPool threadPool = new ThreadPool(settings, executorBuilders.toArray(new ExecutorBuilder[0]));

new ThreadPool对象,从这里开始创建线程池。看懂了ThreadPool类,就理解了ES线程池的一半。

每个操作都有一个线程池,每个线程池都有一个相应的 ExecutorBuilder 对象,线程池都是通过ExecutorBuilder类的build()方法创建的。

在org.elasticsearch.threadpool.ThreadPool.ThreadPool的构建函数里面创建各种ExecutorBuilder对象。可以看出:INDEX操作的线程池的 ExecutorBuilder对象实际类型是FixedExecutorBuilder

 builders.put(Names.GENERIC, new ScalingExecutorBuilder(Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30)));
builders.put(Names.INDEX, new FixedExecutorBuilder(settings, Names.INDEX, availableProcessors, 200, true));
builders.put(Names.WRITE, new FixedExecutorBuilder(settings, Names.WRITE, "bulk", availableProcessors, 200));
builders.put(Names.GET, new FixedExecutorBuilder(settings, Names.GET, availableProcessors, 1000));
builders.put(Names.ANALYZE, new FixedExecutorBuilder(settings, Names.ANALYZE, 1, 16));
builders.put(Names.SEARCH, new AutoQueueAdjustingExecutorBuilder(settings,
Names.SEARCH, searchThreadPoolSize(availableProcessors), 1000, 1000, 1000, 2000));

如上代码所示,虽然ES为我们内置好了许多线程池(GENERIC、INDEX、WRITE、GET...),但还可以自定义 ExecutorBuilder对象,创建自定义的线程池。所有的ExecutorBuilder对象创建完毕后,保存到一个HashMap里面。

        for (final ExecutorBuilder<?> builder : customBuilders) {
if (builders.containsKey(builder.name())) {
throw new IllegalArgumentException("builder with name [" + builder.name() + "] already exists");
}
builders.put(builder.name(), builder);
}

最后,遍历builders 这个HashMap 取出 ExecutorBuilder对象,调用它的build()方法创建线程池

        for (@SuppressWarnings("unchecked") final Map.Entry<String, ExecutorBuilder> entry : builders.entrySet()) {
final ExecutorBuilder.ExecutorSettings executorSettings = entry.getValue().getSettings(settings);
//这里执行build方法创建线程池
final ExecutorHolder executorHolder = entry.getValue().build(executorSettings, threadContext);
if (executors.containsKey(executorHolder.info.getName())) {
throw new IllegalStateException("duplicate executors with name [" + executorHolder.info.getName() + "] registered");
}
logger.debug("created thread pool: {}", entry.getValue().formatInfo(executorHolder.info));
executors.put(entry.getKey(), executorHolder);
}

创建INDEX操作的线程池需要指定任务队列,这个任务队列就是:SizeBlockingQueue。当然了,也有一些其他操作(比如GET操作)的线程池的任务队列也是SizeBlockingQueue。

下面参数可看出:该任务队列的长度为200,org.elasticsearch.threadpool.ThreadPool.ThreadPool的构造方法:

    builders.put(Names.INDEX, new FixedExecutorBuilder(settings, Names.INDEX, availableProcessors, 200, true));

前面已经提到了,每个线程池都由ExecutorBuilder的build方法创建的。具体到INDEX操作的线程池,它的ExecutorBuilder实例对象是: FixedExecutorBuilder对象,在ExecutorBuilder 保存一些线程池参数信息:(core pool size、max pool size、queue size...)

final ExecutorService executor =
EsExecutors.newFixed(settings.nodeName + "/" + name(), size, queueSize, threadFactory, threadContext);

如果queue_size配置为 -1,那就是一个*队列(LinkedTransferQueue)。我们是可以修改线程池配置参数的:关于线程池队列长度的配置信息参考:官方文档threadpool

而INDEX操作对应的线程池的任务队列长度为200,因此下面代码创建了一个长度为200的 SizeBlockingQueue,在代码最后一行,为该线程池指定的拒绝策略是 EsAbortPolicy

    public static EsThreadPoolExecutor newFixed(String name, int size, int queueCapacity, ThreadFactory threadFactory, ThreadContext contextHolder) {
BlockingQueue<Runnable> queue;
if (queueCapacity < 0) {
queue = ConcurrentCollections.newBlockingQueue();
} else {
queue = new SizeBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), queueCapacity);
}
return new EsThreadPoolExecutor(name, size, size, 0, TimeUnit.MILLISECONDS, queue, threadFactory, new EsAbortPolicy(), contextHolder);
}

下面开始分析SizeBlockingQueue的源码:

一般在自定义线程池时,要么是直接 new ThreadPoolExecutor,要么是继承ThreadPoolExecutor,在创建ThreadPoolExecutor对象时需要指定线程池的配置参数。比如,线程池的核心线程数(core pool size),最大线程数,任务队列,拒绝策略。这里我想提一下拒绝策略,因为某些ES的操作具有"强制"执行的特点:如果某个任务被标记为强制执行,那么向线程池提交该任务时,就不能拒绝它。是不是很厉害?想想,线程池是如何做到的?

下面举个例子:

//创建任务队列,这里没有指定任务队列的长度,那么这就是一个*队列
private BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
//创建线程工厂,由它来创建线程
private ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").setUncaughtExceptionHandler(exceptionHandler).build(); //创建线程池,核心线程数为4,最大线程数为16
private ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 16, 1, TimeUnit.DAYS, taskQueue, threadFactory, rejectExecutionHandler);

这里创建的线程池,它的线程数量永远不可能达到最大线程数量16,为什么?因为我们的任务队列是一个*队列,当向线程池中提交任务时,LinkedBlockingQueue.offer方法不会返回false。而在JDK源码java.util.concurrent.ThreadPoolExecutor.execute中,当任务入队列失败返回false时,才有可能触发addWork创建新线程。这个时候,你可能会说:在 new LinkedBlockingQueue的时候指定队列长度不就完了?比如这样指定队列长度为1024

private BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>(1024);

但是,有没有一种方法,能够做到:当core pool size 个核心线程数处理不过来时,先让线程池的线程数量创建到最大值(max pool size),然后,若还有任务提交到线程池,则让任务排队等待处理?SizeBlockingQueue 重写了BlockingQueue的offer方法,实现了这个功能。

另外,我再反问一下?如何确定1024就是一个合适的队列容量?万一提交任务速度很快,一下子任务队列就满了,长度1024就会导致大量的任务被拒绝。

ES中的 ResizableBlockingQueue 实现了一种可动态调整队列长度的任务队列,有兴趣的可以去研究一下。

SizeBlockingQueue 封装了 LinkedTransferQueue,而 LinkedTransferQueue 是一个*队列,与LinkedBlockingQueue不同的是,LinkedTransferQueue的构造方法是不能指定任务队列的长度(capacity)的。因此,SizeBlockingQueue定义一个capacity属性提供了队列有界的功能。

好,来看看SizeBlockingQueue是如何重写offer方法的:org.elasticsearch.common.util.concurrent.SizeBlockingQueue.offer(E)

    @Override
public boolean offer(E e) {
while (true) {
//获取当前任务队列的长度,即:当前任务队列里面有多少个任务正在排队等待执行
final int current = size.get();
//如果正在等待排队的任务数量大于等于任务队列长度的最大值(容量),
//返回false 就有可能 触发 java.util.concurrent.ThreadPoolExecutor.addWorker 调用创建新线程
if (current >= capacity()) {
return false;
} //当前正在排队的任务数量尚未超过队列的最大长度,使用CAS 先将任务队列长度加1,[CAS的经典用法]
if (size.compareAndSet(current, 1 + current)) {
break;
}
}
//将任务添加到队列
boolean offered = queue.offer(e);
if (!offered) {
//如果未添加成功,再把数量减回去即可
size.decrementAndGet();
}
return offered;
}

上面,就是通过先判断当前排队的任务是否小于任务队列的最大长度(容量) 来实现:优先创建线程数量到 max pool size。下面来模拟一下使用 SizeBlockingQueue 时处理任务的步骤:

根据前面的介绍:线程池 core pool size=4,max pool size=16,taskQueue 是 SizeBlockingQueue,任务队列的最大长度是200

1,提交1-4号 四个任务给线程池,线程池创建4个线程处理这些任务

2,1-4号 四个任务正在执行中...此时又提交了8个任务到线程池

3,这时,线程池是再继续创建8个线程,处理 5-12号任务。此时,线程池中一共有4+8=12个线程,小于max pool size

4,假设 1-12号任务都正在处理中,此时又提交了8个任务到线程池

5,这时,线程池会再创建4个新线程处理其中的13-16号 这4个任务,线程数量已经达到max pool size,不能再创建新线程了,还有4个任务(17-20号)入队列排队等待。

有没有兴趣模拟一下使用LinkedBlockingQueue作为任务队列时,线程池又是如何处理这一共提交的20个任务的?

最后来分析一下 SizeBlockingQueue 如何支持:当向线程池提交任务时,如果任务被某种拒绝策略拒绝了,如果这种任务又很重要,那能不能强制将该任务提交到线程池的任务队列中呢?

这里就涉及到:在创建线程池时,为线程池配置了何种拒绝策略了。下面以INDEX操作的线程池为例说明:

在org.elasticsearch.common.util.concurrent.EsExecutors.newFixed 中:可知该线程池所使用的拒绝策略是:EsAbortPolicy

return new EsThreadPoolExecutor(name, size, size, 0, TimeUnit.MILLISECONDS, queue, threadFactory, new EsAbortPolicy(), contextHolder);

看 EsAbortPolicy 的源码:org.elasticsearch.common.util.concurrent.EsAbortPolicy.rejectedExecution

if (r instanceof AbstractRunnable) {
//判断该任务是不是一个 可强制提交的任务
if (((AbstractRunnable) r).isForceExecution()) {
BlockingQueue<Runnable> queue = executor.getQueue();
if (!(queue instanceof SizeBlockingQueue)) {
throw new IllegalStateException("forced execution, but expected a size queue");
}
//是一个可强制提交的任务,并且 线程池的任务队列是 SizeBlockingQueue时,强制提交任务
try {
((SizeBlockingQueue) queue).forcePut(r);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("forced execution, but got interrupted", e);
}
return;
}
}
rejected.inc();
//任务被拒绝且未能强制执行, 抛出EsRejectedExecutionException异常后,会被 EsThreadPoolExecutor.doExecute catch, 进行相应的处理
throw new EsRejectedExecutionException("rejected execution of " + r + " on " + executor, executor.isShutdown());

AbstractRunnable 是提交的Runnable任务,只要Runnable任务的 isForceExecution()返回true,就表明这个任务需要“强制提交”。关于AbstractRunnable,可参考:Elasticsearch中各种线程池分析

那为什么只有当任务队列是 SizeBlockingQueue 时,才可以强制提交呢?这很好理解:首先SizeBlockingQueue它封装了LinkedTransferQueue,LinkedTransferQueue本质上是一个*队列,实际上可以添加无穷多个任务(不考虑OOM),只不过是用 capacity 属性限制了队列的长度而已。

如果,任务队列是 new LinkedBlockingQueue<>(1024),肯定是不能支持强制提交的,因为当LinkedBlockingQueue长度达到1024后,再提交任务,直接返回false了。从这里也可以借鉴ES线程池任务队列的设计方式,应用到项目中去。

综上:只有Runnable任务 isForceExecution返回true,并且线程池的任务队列是SizeBlockingQueue时,向线程池提交任务时,总是能提交成功(强制执行机制保证)。其他情况下,任务被拒绝时,会抛出EsRejectedExecutionException异常。

强制提交,把任务添加到任务队列 SizeBlockingQueue 中,源码如下:

org.elasticsearch.common.util.concurrent.SizeBlockingQueue.forcePut

    /**
* Forces adding an element to the queue, without doing size checks.
*/
public void forcePut(E e) throws InterruptedException {
size.incrementAndGet();
try {
queue.put(e);
} catch (InterruptedException ie) {
size.decrementAndGet();
throw ie;
}
}

总结:

ES会为每种操作创建一个线程池,本文基于INDEX操作分析了ES中线程池的任务队列SizeBlockingQueue。对于 INDEX 操作而言,它的线程池是由org.elasticsearch.threadpool.FixedExecutorBuilder 的build方法创建的,线程池的最大核心线程数和最大线程数相同,使用的任务队列是 SizeBlockingQueue,长度为200,拒绝策略是:org.elasticsearch.common.util.concurrent.EsAbortPolicy。

为什么要为不同的操作分配不同的线程池呢?

假设 index 操作 和 snapshot 操作使用同一个线程池,如果某节点发生故障,index操作被阻塞了,而 Client发起的索引文档操作的 QPS又很高,就很容易影响 snapshot 服务了。

SizeBlockingQueue 本质上是一个 LinkedTransferQueue,其实ES中所有的任务队列都是封装LinkedTransferQueue实现的,并没有使用LinkedBlockingQueue。

ES中的所有任务(Runnable)都是基于org.elasticsearch.common.util.concurrent.AbstractRunnable这个抽象类封装的,当然有一些任务是通过Lambda表达式的形式提交的。任务的具体处理逻辑在 org.elasticsearch.common.util.concurrent.AbstractRunnable#doRun 方法中,任务执行完成由onAfter()处理,执行出现异常由onFailure()处理。线程池的 org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor#doExecute 方法 里面就是整个任务的处理流程:

    protected void doExecute(final Runnable command) {
try {
super.execute(command);
} catch (EsRejectedExecutionException ex) {
if (command instanceof AbstractRunnable) {
// If we are an abstract runnable we can handle the rejection
// directly and don't need to rethrow it.
try {
((AbstractRunnable) command).onRejection(ex);
} finally {
((AbstractRunnable) command).onAfter(); }
} else {
throw ex;
}
}
}

最后,ES的线程池模块代码主要在 org.elasticsearch.threadpool 和 org.elasticsearch.common.util.concurrent 包下。总体来说,threadpool模块相比于ES的其他模块,是一个小模块,代码不算复杂。但是threadpool又很重要,因为它是其他模块执行逻辑的基础,threadpool 再配上异步执行机制,是ES源码中其他操作的源码实现思路。

ElasticSearch 线程池类型分析之SizeBlockingQueue

ElasticSearch 线程池类型分析之SizeBlockingQueue

参考:

探究ElasticSearch中的线程池实现

Elasticsearch中各种线程池分析

上一篇:CVE-2019-0797漏洞:Windows操作系统中的新零日在攻击中被利用


下一篇:iOS UImage 与 RGB 裸数据的相互转换