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。
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中的线程池实现
Elasticsearch中各种线程池分析

原文地址:https://www.cnblogs.com/hapjin/p/10999788.html

时间: 2024-11-05 12:14:39

ElasticSearch 线程池类型分析之SizeBlockingQueue的相关文章

JAVA线程池的分析和使用

http://www.infoq.com/cn/articles/java-threadPool/ 1. 引言 合理利用线程池能够带来三个好处.第一:降低资源消耗.通过重复利用已创建的线程降低线程创建和销毁造成的消耗.第二:提高响应速度.当任务到达时,任务可以不需要等到线程创建就能立即执行.第三:提高线程的可管理性.线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控.但是要做到合理的利用线程池,必须对其原理了如指掌. 2. 线程池

AsyncTask中线程池调度分析

尊重原创:http://blog.csdn.net/yuanzeyao/article/details/42583215 在Android中,和异步执行相关的两个类就是Handler和AsyncTask,所以Android开发人员对于这两个类是再熟悉不过了,所以这里我不是讲解AsyncTask怎么使用,而是想分析一下AsyncTask中线程池的调度过程,然后简单的介绍一下AsyncTask的源码以及Android3.0前后,AsyncTask中线程池的区别. 在正式学习AsyncTask中的线程

ThreadPoolExecutor线程池的分析和使用

1. 引言 合理利用线程池能够带来三个好处. 第一:降低资源消耗.通过重复利用已创建的线程降低线程创建和销毁造成的消耗. 第二:提高响应速度.当任务到达时,任务可以不需要等到线程创建就能立即执行. 第三:提高线程的可管理性.线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控. 但是要做到合理的利用线程池,必须对其原理了如指掌. 2. 线程池的使用 线程池的创建 我们可以通过java.util.concurrent.ThreadPo

[转]JAVA线程池的分析和使用

1. 引言 合理利用线程池能够带来三个好处. 第一:降低资源消耗.通过重复利用已创建的线程降低线程创建和销毁造成的消耗. 第二:提高响应速度.当任务到达时,任务可以不需要等到线程创建就能立即执行. 第三:提高线程的可管理性.线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控. 但是要做到合理的利用线程池,必须对其原理了如指掌. 2. 线程池的使用 线程池的创建 我们可以通过ThreadPoolExecutor来创建一个线程池. n

java并发包&amp;线程池原理分析&amp;锁的深度化

      java并发包&线程池原理分析&锁的深度化 并发包 同步容器类 Vector与ArrayList区别 1.ArrayList是最常用的List实现类,内部是通过数组实现的,它允许对元素进行快速随机访问.数组的缺点是每个元素之间不能有间隔,当数组大小不满足时需要增加存储能力,就要讲已经有数组的数据复制到新的存储空间中.当从ArrayList的中间位置插入或者删除元素时,需要对数组进行复制.移动.代价比较高.因此,它适合随机查找和遍历,不适合插入和删除. 2.Vector与Arra

聊聊并发(三)Java线程池的分析和使用

引言 合理利用线程池能够带来三个好处.第一:降低资源消耗.通过重复利用已创建的线程降低线程创建和销毁造成的消耗.第二:提高响应速度.当任务到达时,任务可以不需要的等到线程创建就能立即执行.第三:提高线程的可管理性.线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控.但是要做到合理的利用线程池,必须对其原理了如指掌. 线程池的使用 线程池的创建 我们可以通过ThreadPoolExecutor来创建一个线程池. new Thread

Java 线程池原理分析

1.简介 线程池可以简单看做是一组线程的集合,通过使用线程池,我们可以方便的复用线程,避免了频繁创建和销毁线程所带来的开销.在应用上,线程池可应用在后端相关服务中.比如 Web 服务器,数据库服务器等.以 Web 服务器为例,假如 Web 服务器会收到大量短时的 HTTP 请求,如果此时我们简单的为每个 HTTP 请求创建一个处理线程,那么服务器的资源将会很快被耗尽.当然我们也可以自己去管理并复用已创建的线程,以限制资源的消耗量,但这样会使用程序的逻辑变复杂.好在,幸运的是,我们不必那样做.在

线程池ThreadPoolExecutor分析

线程池.线程池是什么,说究竟,线程池是处理多线程的一种形式,管理线程的创建,任务的运行,避免了无限创建新的线程带来的资源消耗,可以提高应用的性能.非常多相关操作都是离不开的线程池的,比方android应用中网络请求的封装.这篇博客要解决的问题是: 1.线程池的工作原理及过程. 要分析线程池的工作原理及过程,还是要从它的源代码实现入手,首先是线程是构造方法,何谓构造方法.构造方法就是对成员变量进行初始化,在这里,我们能够看到它的构造方法: /** * Creates a new {@code Th

线程池ThreadPoolExecutor分析: 线程池是什么时候创建线程的,队列中的任务是什么时候取出来的?

带着几个问题进入源码分析: 线程池是什么时候创建线程的? 任务runnable task是先放到core到maxThread之间的线程,还是先放到队列? 队列中的任务是什么时候取出来的? 什么时候会触发reject策略? core到maxThread之间的线程什么时候会die? task抛出异常,线程池中这个work thread还能运行其他任务吗? 至少在new ThreadPoolExecutor()时,Thread对象并没有初始化. 这里仅仅指定了几个初始参数 一段基础代码,进入分析 pu