《Elasticsearch 源码解析与优化实战》第16章：ThreadPool模块分析

一、简介

每个节点都会创建一系列的线程池来执行任务，许多线程池都有与其相关任务队列，用来允许挂起请求，而不是丢弃它。 下面列出目前ES版本中的线程池。

官网： https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-threadpool.html

• generic：用于通用的操作(例如，节点发现)，线程池类型为 scaling。
• index：用于index/delete操作，线程池类型为fixed，大小为处理器的数量，队列大小为200，允许设置的最大线程数为1+处理器数量。
• search：用于count/search/suggest操作。线程池类型为fixed, 大小为int((处理器数量3)/2)+1，队列大小为1000。
• get：用于get操作。线程池类型为fixed, 大小为处理器的数量，队列大小为1000。
• bulk：用于bulk操作，线程池类型为fixed,大小为处理器的数量，队列大小为200，该线程池允许设置的最大线程数为1+处理器数量。
• snapshot：用于snaphostrestore操作。线程池类型为scaling,线程保持存活时间为5min,最大线程数为min(5, (处理器数量)/2)。
• warme：用于segment warm-up操作。线程池类型为scaling， 线程保持存活时间为5min，最大线程数为min(5, (处理器数量)/2)。
• refresh：用于refresh 操作。线程池类型为scaling, 线程空闲保持存活时间为5min，最大线程数为min(10, (处理器数量)/2)。
• listener：主要用于Java客户端线程监听器被设置为true时执行动作。线程池类型为scaling,最大线程数为min(10, (处理器数量)/2)。
• same：在调用者线程执行，不转移到新的线程池。
• management：管理工作的线程池，例如，Node info、Node tats、 List tasks等。
• flush：用于索引数据的flush操作。
• force_merge：顾名思义，用于Lucene分段的force merge。
• fetch_shard_started ：用于TransportNodesAction.
• fetch_shard_store ：用于TransportNodesListShardStoreMetaData。
• thread_pool.search.size: 30：线程池和队列的大小可以通过配置文件进行调整,例如，为search增加线程数和队列大小:

二、线程池类型

如同任何要并发处理任务的服务程序一样，线程池要处理的任务类型大致可以分为两类:CPU计算密集型和I/O密集型。对于两种不同的任务类型,需要为线程池设置不同的线程数量。

一般说来，线程池的大小可以参考如下设置，其中N为CPU的个数:

• 对于CPU密集型任务，线程池大小设置为N+1;
• 对于I/O密集型任务，线程池大小设置为2N+1；

对于计算密集型任务，线程池的线程数量 一 般不应该超过N+1。如果线程数量太多，则会导致更高的线程间上下文切换的代价。加1是为了当计算线程出现偶尔的故障,或者偶尔的I/O、发送数据、写日志等情况时，这个额外的线程可以保证CPU时钟周期不被浪费。

I/O密集型任务的线程数可以稍大一些，因为I/O密集型任务大部分时间阻塞在I/O过程，适当增加线程数可以增加并发处理能力。而上下文切换的代价相对来说已经不那么敏感。但是线程数量不一定设置为2N+1,具体需要看I/O等待时间有多长。等待时间越长，需要越多的线程，等待时间越少，需要越少的线程。因此也可以参考下面的公式:

最佳线程数= ((线程等待时间 + 线程CPU时间) /线程CPU时间) * CPU数

为了应对这两种类型的任务，ES封装了以下类型的线程池。

2.1、fixed

线程池拥有固定数量的线程来处理请求，当线程空闲时不会销毁，当所有线程都繁忙时，请求被添加到队列中。

• size参数用来控制线程的数量。
• queue_size 参数用来控制线程池相关的任务队列大小。设置为-1表示无限制。当请求到达时，如果队列已满，则请求将被拒绝。

例如:

thread_pool.search.size: 30
thread_pool.search.queue_size: 1500

2.2、scaling

scaling 线程池的线程数量是动态的，介于core和max参数之间变化。线程池的最小线程数为配置的core大小，随着请求的增加，当core数量的线程全都繁忙时，线程数逐渐增大到max数量。max是线程池可拥有的线程数.上限。当线程空闲时，线程数从max大小逐渐降低到core大小。

• keep_alive参数用来控制线程在线程池中的最长空闲时间。

例如：

thread_pool.warmer.core: 1
thread_pool.warmer.max:8
thread_pool.warmer.keep_alive: 2m 

2.3、direct

这种线程池对用户并不可见,当某个任务不需要在独立的线程执行，又想被线程池管理时，于是诞生了这种特殊类型的线程池:在调用者线程中执行任务。

2.4、fixed_auto_queue_size

与fixed类型的线程池相似，该线程池的线程数量为固定值，但是队列类型不一样。 其队列大小根据利特尔法则( Little’s Law) 自动调整大小。该法则的详细信息可以参考https://en.wikipedia.org/wiki/Little%27s_law。 该线程池有以下参数可以调整:

• size：用于指定线程数量;
• queue_size,：指定初始队列大小; .
• min_queue_size：最小队列大小;
• max_queue_size：最大队列大小;
• auto_gueue_frame_size： 调整队列之前进行测量的操作数;
• target_response_time：一个时间值设置，用来指示任务的平均响应时间，如果任务经常高于这个时间，则线程池队列将被调小，以便拒绝任务。

该线程类型为实验性质，未来可能会移除。目前只有search线程池使用这种类型。（ Deprecated in 7.7.0 and will be removed in 8.0. ）

三、处理器设置

https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-threadpool.html#node.processors

默认情况下，ES自动探测处理器数量。各个线程池的大小基于这个数量进行初始化。在某些情况下，如果想手工指定处理器数量，则可以通过设置 processors 参数实现：

processors: 2

有以下几种场景是需要明确设置processors数量的：

(1)、在同一台主机上运行多个ES实例，但希望每个实例的线程池只根据一部分CPU来设置，此时可以通过processors参数来设置处理器数量。例如，在16核的服务器上运行2个实例，可以将processors设置为8。请注意，在单台主机上运行多个实例，除了设置processors数量，还有许多更复杂的参数需要设置。例如，修改GC线程数，绑定进程到CPU等。

(2)、有时候自动探测出的处理器数量是错误的，在这种情况下，需要明确设置处理器数量。要检查自动探测的处理器数量，可以使用节点信息API中的os字段来查看。

四、查看线程池

ES提供了丰富的API查看线程池状态，在监控节点健康、排查问题时非常有用。

4.1、cat thread pool

该命令显示每个节点的线程池统计信息。默认情况下，所有线程池都返回活动、队列和被拒绝的统计信息。我们需要特别留意被拒接的信息，例如，bulk 请求被拒绝意味着客户端写入失败。在正常情况下客户端应该捕获这种错误(错误码429)并延迟重试，但有时客户端不一定对这种错误做了处理，导致写入集群的数据量低于预期值。

curl -X GET "localhost:9200/_cat/thread_pool"

返回信息如下：

• active 表示当前正在执行任务的线程数量
• queue 表示队列中等待处理的请求数量
• rejected 表示由于队列已满，请求被拒绝的次数。

对返回结果进行过滤等更多用法可参考官方手册: https://www.elastic.co/guide/en/elasticsearch/reference/6.1/cat-thread-pool.html。

4.2、nodes info

节点信息API可以返回每个线程池的类型和配置信息，例如，线程数量、队列大小等。下面的第一条命令返回所有节点的信息，第二条命令返回特定节点的信息。.

curl -X GET "localhost:9200/_nodes"
curl -X GET "localhost:9200/_nodes/nodeId1, nodeId2"

节点信息API返回的信息非常大，其中与线程池相关信息在thread_ pool 字段中，选取部分信息如下:

"thread_pool" : {
    "force_merge" : {
        "type" : "fixed",
        "min" : 1,
        "max" : 1,
        "queue_size" : - 1
    },
    "fetch_shard started" : {
        "type" : "scaling",
        "min" : 1,
        "max" : 16,
        "keep_alive" : "5m",
        "queue_size" : -1
    }
}

该命令的完整信息可参考官方手册: https://www.elastic.co/guide/en/elasticsearch/reference/6.1/cluster-nodes-info.html

4.3、nodes stats

statsAPI返回集群中一个或全部节点的统计数据。

下面的第一条命令返回所有节点的统计数据，第二条命令返回特定节点的统计数据。

curl -X GET “localhost:9200/_nodes/stats” curl -X GET “localhost:9200/_nodes/nodeId1, nodeId2/stats”

默认情况下，该API返回全部 indices、oS、process、jvm、transport、http、fs、breaker 和thread_pool 方面的统计数据。其中线程池相关的返回结果摘要如下:

"thread_pool" : {
    "bulk" : {
        "threads" : 0,
        "queue" : 0,
        "active" : 0,
        "rejected" : 0，
        "largest" : 0，
        "completed" : 0
    }
}

该命令的完整使用方式可参考官方手册: https://www.elastic.co/guide/en/elasticsearch/reference/6.1/cluster-nodes-stats.html。

4.4、nodes hot threads

该API返回集群中一个或全部节点的热点线程。

当发现节点进程占用CPU非常高时，想知道是哪些线程导致的，这些线程具体在执行什么操作，常规做法是通过top命令配合jstack来定位线程，现在ES提供了更便捷的方式，通过hot threads API可以直接返回这些信息。

下面的第一条命令返回所有节点的热点线程，第二条命令返回特定节点的热点线程。

curl -X GET "localhost:9200/_nodes/hot_threads"
curl -X GET "localhost:9200/nodes/nodeId1, nodeId2/hot_threads"

该命令支持以下参数：

• threads：返回的热点线程数，默认为3。
• interval：ES对线程做两次检查,来计算某个操作.上花费时间的百分比，此参数定义这个间隔时间。默认为500 ms。
• type：定义要检查的线程状态类型,默认为CPU.API可以检查线程的CPU占用时间、阻塞(block)时间和等待(wait) 时间。
• ignore_idle_threads：如果设置为true， 则空闲线程(例如，在套接字中等待，或者从空队列中获取任务)将被过滤。默认值为true。

其返回信息的样例如下图所示。

返回信息中的第一行表明这个是哪个节点的信息，以及这个节点的IP地址等。

::: {node1} {un- 9UZ4PS8-K6hF59x1MWA} {bjk2C_ _6USh0YgMYKcBWKLQ} {node1.eshost}
{10.10.13.15:9300}

接下来列出哪个线程占用较多的CPU，以及CPU的占用比：

82.2% (411.2ms out of 500ms) cpu usage by thread 'elasticsearch [node1]
[bulk] [T#1] '

最后是该线程的堆栈信息。

ES中的线程池是基于对Java线程池的封装和扩展。我们先看一下Java线程池的结构和使用方式，这些是ES内部线程原理的基础知识。

4.5、Java 的线程池结构

Java内部的线程池称为Executor框架，几个基本的类概念如下：

• Runable定义一个要执行的任务。
• Executor提供了execute 方法，接受一个Runable实例，用来执行一个任务。
• ExecutorService是线程池的虚基类，继承自Executor, 提供了shutdown, shutdownNow等关闭线程池接口。
• ThreadPoolExecutor线程池的具体实现。继承自ExecutorService，维护线程创建、线程生命周期、任务队列等。
• EsThreadPoolExecutor是ES对ThreadPoolExecutor的扩展实现。未来会增加一些统计信息。

这几个类的继承结构如下图所示。

我们以一个简单的例子来看看Java 线程池的用法，ExecutorService 类用于保存创建的线程池实例，后续调用execute方法执行任务。在下面的例子中，任务类TestRunnable只是打印当前线程名称。

import java. util.concurrent.ExecutorService;
import java. util.concurrent.Executors;
public class ThreadPoolDemo {
    public static void main(String[] args) {
        //通过Executors构建一个固定大小的线程池，线程数量为2，返回线程池实例
        ExecutorService executorService = Executors.newFixedThreadPool (2) ;
        //调用 线程池的execute方法执行一个任务
        executorService.execute(new TestRunnable());
    }
}
class TestRunnable implements Runnable {
    public void run() {
        System.out.println (Thread.currentThread().getName()) ;
    }
}

ES内部创建线程池时，返回类型同样是ExecutorService类。接下来我们通过构建过程来看看ThreadPoolExecutor的结构，其构造函数如下:

public ThreadPoolExecutor (int corePoolSize,
                int maximumPoolSize,
                        long keepAl iveTime,
                TimeUnit unit,
                BlockingQueue workQueue,
                ThreadFactory threadFactory,
                RejectedExecutionHandler handler) {
        if (corePoolSize