Elasticsearch源码分析-架构设计之Action

Posted 2023-05-05

参考技术A 如果我们发送如下的http请求，elasticsearch是如何匹配对应Action的呢？

在org.elasticsearch.rest.action包下面，存放了处理各种请求的rest action类
在每个action的构造方法中，会将当前action对象注册到指定的url上

在registerHandler()方法中，会将url path和对应handler添加到http method对象上

elasticsearch使用HttpRequestHandler接收http请求消息，最终会在executeHandler()方法中调用getHandler()方法获取path对应的handler

在getHandler()方法中，主要是根据method和path获取在RestXxxAction中添加的handler

elasticsearch在path对应获取Handler后，就执行该handler的handleRequest()方法，作为rest逻辑入口

在ActionModule中，会将Action和对应的TransportAction注册到actions对象中

在RestXxxAction中，最终会调用AbstractClient的xxx方法，并传入Action参数为XxxAction.INSTANCE，即为在ActionModule中注册的Action类

在NodeClient中会根据传入的Action参数从actions获取在ActionModule中注册的TransportAction，并执行其execute()方法

在TransportAction类图个execute()方法中可以看出：
① TransportXxxAction都继承自TransportAction类
② TransportAction的execute()方法仅调用了doExecute()方法
④ TransportXxxAction重写了TransportAction的doExecute()方法
因此，在最终的对应关系为在ActionModule中注册的Action调用的为TransportXxxAction的doExecute()方法

TransportAction的execute()方法仅调用了需要子类重写的抽象方法doExecute()

如果elasticsearch要请求的节点不是当前节点，则需要将请求发送到对应的节点上执行
在TransportXxxAction中会将ACTION和对应的Handler注册到对应的serverHandlers对象中，在使用transportService发送请求后，MessageChannelHandler.messageReceived()会接受到信息，然后在handleRequest()方法中获取注册的Handler，执行其messageReceived()方法或者交给线程池处理

在transportServiceAdapter.handler()方法中，即根据注册的action获取对应的TransportHandler，如果executor是same则执行handler的messageReceived()方法，否则交给线程池执行

下面讲另外一种情况，在ActionModule中注册的Action也都会注册到handler中

在TransportAction类中同样使用transportService.registerHandler()方法注册handler

因此在InternalTransportClient中，主要是回调proxy.execute()

transportService.sendRequest()中的action.name()即为Action的Name参数，已被注册到对应的handler中了

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中各种线程池分析