Elasticsearch传输客户端产生大量线程

From an Apache storm bolt, I am using Elasticsearch's transport client to remotely connect to an ES cluster. When I take a jstack output of the storm process, I notice that there are nearly 1000 threads with an ES stack trace like:

从Apache风暴螺栓,我使用Elasticsearch的传输客户端远程连接到ES集群。当我接受风暴过程的jstack输出时,我注意到有近1000个具有ES堆栈跟踪的线程,如:

elasticsearch[Flying Tiger][transport_client_worker][T#22]{New I/O worker #269}" daemon prio=10 tid=0x00007f80ac3cb000 nid=0x356b runnable [0x00007f7e96b2a000]
   java.lang.Thread.State: RUNNABLE
    at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
    at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
    at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:79)
    at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:87)
    - locked <0x00000000d148d138> (a sun.nio.ch.Util$2)
    - locked <0x00000000d148d128> (a java.util.Collections$UnmodifiableSet)
    - locked <0x00000000d148c9b8> (a sun.nio.ch.EPollSelectorImpl)
    at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:98)
    at org.elasticsearch.common.netty.channel.socket.nio.SelectorUtil.select(SelectorUtil.java:68)
    at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioSelector.select(AbstractNioSelector.java:415)
    at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:212)
    at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89)
    at org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178)
    at org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
    at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

I am using a single instance of the ES transport client across my storm topology, which has about 18 output streams invoking the ES client to write data to the ES cluster.

我在我的风暴拓扑中使用了一个ES传输客户端实例,它有大约18个输出流调用ES客户端将数据写入ES集群。

Why does the ES transport client spawn so many threads ? Is there any way I can tune this ? I tried looking up ES documentation but it does not provide any internal details on the threading mechanism of the transport client nor does it give an option to tune the number of threads of the client.

为什么ES传输客户端会产生这么多线程?有什么方法可以调整这个吗?我尝试查找ES文档,但它没有提供有关传输客户端的线程机制的任何内部详细信息,也没有提供调整客户端线程数的选项。

2 个解决方案

#1

I had very similar experience before. As you've mentioned, one transport client creates tens of threads including timers and so on.

我之前有过非常相似的经历。正如您所提到的,一个传输客户端会创建数十个线程,包括计时器等。

What you have to check is whether there is exactly one transport client per worker process. Back in my earlier days, when I used 32 transport clients, there were more than 1 thousand threads and after I've correctly modified it to be singleton instance, number of threads decreased to less than 2 hundreds(including all other threads created in my topology)

您需要检查的是每个工作进程是否只有一个传输客户端。回到我早期,当我使用32个传输客户端时,有超过1千个线程,在我正确地将其修改为单例实例后,线程数减少到不到2百(包括我创建的所有其他线程)拓扑)

#2

You can define system property: "es.processors.override" or setting "processors", based on source code of org.elasticsearch.common.util.concurrent.EsExecutors. I tried this method and limit the number of worker threads successfully.

您可以根据org.elasticsearch.common.util.concurrent.EsExecutors的源代码定义系统属性:“es.processors.override”或设置“处理器”。我尝试了这种方法并成功限制了工作线程的数量。

/**
 * Settings key to manually set the number of available processors.
 * This is used to adjust thread pools sizes etc. per node.
 */
public static final String PROCESSORS = "processors";

/** Useful for testing */
public static final String DEFAULT_SYSPROP = "es.processors.override";

Also from limit number of thread in ThreadPool while creating TransportClient in elasticsearch

同时也是在elasticsearch中创建TransportClient时ThreadPool中的线程限制数

Settings settings = ImmutableSettings.settingsBuilder()
                    .put("transport.netty.workerCount",NUM_THREADS)
                    .build();

#1