文档章节

聊聊Elasticsearch的EsThreadPoolExecutor

go4it
 go4it
发布于 06/02 11:51
字数 1417
阅读 20
收藏 1

本文主要研究一下Elasticsearch的EsThreadPoolExecutor

EsThreadPoolExecutor

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java

public class EsThreadPoolExecutor extends ThreadPoolExecutor {

    private final ThreadContext contextHolder;
    private volatile ShutdownListener listener;

    private final Object monitor = new Object();
    /**
     * Name used in error reporting.
     */
    private final String name;

    final String getName() {
        return name;
    }

    EsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, ThreadContext contextHolder) {
        this(name, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new EsAbortPolicy(), contextHolder);
    }

    @SuppressForbidden(reason = "properly rethrowing errors, see EsExecutors.rethrowErrors")
    EsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, XRejectedExecutionHandler handler,
            ThreadContext contextHolder) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.name = name;
        this.contextHolder = contextHolder;
    }

    public void shutdown(ShutdownListener listener) {
        synchronized (monitor) {
            if (this.listener != null) {
                throw new IllegalStateException("Shutdown was already called on this thread pool");
            }
            if (isTerminated()) {
                listener.onTerminated();
            } else {
                this.listener = listener;
            }
        }
        shutdown();
    }

    @Override
    protected synchronized void terminated() {
        super.terminated();
        synchronized (monitor) {
            if (listener != null) {
                try {
                    listener.onTerminated();
                } finally {
                    listener = null;
                }
            }
        }
    }

    public interface ShutdownListener {
        void onTerminated();
    }

    @Override
    public void execute(Runnable command) {
        command = wrapRunnable(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;
            }
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        EsExecutors.rethrowErrors(unwrap(r));
        assert assertDefaultContext(r);
    }

    private boolean assertDefaultContext(Runnable r) {
        try {
            assert contextHolder.isDefaultContext() : "the thread context is not the default context and the thread [" +
                Thread.currentThread().getName() + "] is being returned to the pool after executing [" + r + "]";
        } catch (IllegalStateException ex) {
            // sometimes we execute on a closed context and isDefaultContext doen't bypass the ensureOpen checks
            // this must not trigger an exception here since we only assert if the default is restored and
            // we don't really care if we are closed
            if (contextHolder.isClosed() == false) {
                throw ex;
            }
        }
        return true;
    }

    /**
     * Returns a stream of all pending tasks. This is similar to {@link #getQueue()} but will expose the originally submitted
     * {@link Runnable} instances rather than potentially wrapped ones.
     */
    public Stream<Runnable> getTasks() {
        return this.getQueue().stream().map(this::unwrap);
    }

    @Override
    public final String toString() {
        StringBuilder b = new StringBuilder();
        b.append(getClass().getSimpleName()).append('[');
        b.append("name = ").append(name).append(", ");
        if (getQueue() instanceof SizeBlockingQueue) {
            @SuppressWarnings("rawtypes")
            SizeBlockingQueue queue = (SizeBlockingQueue) getQueue();
            b.append("queue capacity = ").append(queue.capacity()).append(", ");
        }
        appendThreadPoolExecutorDetails(b);
        /*
         * ThreadPoolExecutor has some nice information in its toString but we
         * can't get at it easily without just getting the toString.
         */
        b.append(super.toString()).append(']');
        return b.toString();
    }

    /**
     * Append details about this thread pool to the specified {@link StringBuilder}. All details should be appended as key/value pairs in
     * the form "%s = %s, "
     *
     * @param sb the {@link StringBuilder} to append to
     */
    protected void appendThreadPoolExecutorDetails(final StringBuilder sb) {

    }

    protected Runnable wrapRunnable(Runnable command) {
        return contextHolder.preserveContext(command);
    }

    protected Runnable unwrap(Runnable runnable) {
        return contextHolder.unwrap(runnable);
    }
}
  • EsThreadPoolExecutor继承了ThreadPoolExecutor,它提供了两个构造器,它们要求RejectedExecutionHandler为XRejectedExecutionHandler类型,其中一个构造器默认为EsAbortPolicy,它们还要求传入ThreadContext
  • 它覆盖了terminated、execute、afterExecute方法,其中terminated方法会回调listener.onTerminated();execute方法会捕获EsRejectedExecutionException异常,在command为AbstractRunnable类型时回调其onRejection及onAfter方法;afterExecute方法会执行EsExecutors.rethrowErrors(unwrap(r))方法
  • 它提供了wrapRunnable及unwrap方法,分别会调用contextHolder.preserveContext及contextHolder.unwrap方法

XRejectedExecutionHandler

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/XRejectedExecutionHandler.java

public interface XRejectedExecutionHandler extends RejectedExecutionHandler {

    /**
     * The number of rejected executions.
     */
    long rejected();
}
  • XRejectedExecutionHandler接口继承了RejectedExecutionHandler接口,它定义了rejected方法返回rejected的数量;它有两个实现类分别为EsAbortPolicy及ForceQueuePolicy

EsAbortPolicy

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/EsAbortPolicy.java

public class EsAbortPolicy implements XRejectedExecutionHandler {
    private final CounterMetric rejected = new CounterMetric();

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        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");
                }
                try {
                    ((SizeBlockingQueue) queue).forcePut(r);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new IllegalStateException("forced execution, but got interrupted", e);
                }
                return;
            }
        }
        rejected.inc();
        throw new EsRejectedExecutionException("rejected execution of " + r + " on " + executor, executor.isShutdown());
    }

    @Override
    public long rejected() {
        return rejected.count();
    }
}
  • EsAbortPolicy实现了XRejectedExecutionHandler接口,其内部使用CounterMetric类维护rejected数量,而rejected方法直接返回该值;rejectedExecution方法对AbstractRunnable类型的runnable会判断是否isForceExecution,且是SizeBlockingQueue,则调用SizeBlockingQueue的forcePut方法重新force执行该runnable,之后就是递增rejected计数

ForceQueuePolicy

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java

    static class ForceQueuePolicy implements XRejectedExecutionHandler {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                // force queue policy should only be used with a scaling queue
                assert executor.getQueue() instanceof ExecutorScalingQueue;
                executor.getQueue().put(r);
            } catch (final InterruptedException e) {
                // a scaling queue never blocks so a put to it can never be interrupted
                throw new AssertionError(e);
            }
        }

        @Override
        public long rejected() {
            return 0;
        }

    }
  • ForceQueuePolicy实现了XRejectedExecutionHandler接口,它的rejectedExecution方法仅仅对ExecutorScalingQueue进行重新入队操作,而rejected方法返回0

AbstractRunnable

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/AbstractRunnable.java

public abstract class AbstractRunnable implements Runnable {

    /**
     * Should the runnable force its execution in case it gets rejected?
     */
    public boolean isForceExecution() {
        return false;
    }

    @Override
    public final void run() {
        try {
            doRun();
        } catch (Exception t) {
            onFailure(t);
        } finally {
            onAfter();
        }
    }

    /**
     * This method is called in a finally block after successful execution
     * or on a rejection.
     */
    public void onAfter() {
        // nothing by default
    }

    /**
     * This method is invoked for all exception thrown by {@link #doRun()}
     */
    public abstract void onFailure(Exception e);

    /**
     * This should be executed if the thread-pool executing this action rejected the execution.
     * The default implementation forwards to {@link #onFailure(Exception)}
     */
    public void onRejection(Exception e) {
        onFailure(e);
    }

    /**
     * This method has the same semantics as {@link Runnable#run()}
     * @throws InterruptedException if the run method throws an InterruptedException
     */
    protected abstract void doRun() throws Exception;
}
  • AbstractRunnable声明实现Runnable接口,它的run方法分别会回调doRun、onFailure、onAfter方法;另外它还定义了isForceExecution方法用于确定当rejected的时候是否force execution

小结

  • EsThreadPoolExecutor继承了ThreadPoolExecutor,它提供了两个构造器,它们要求RejectedExecutionHandler为XRejectedExecutionHandler类型,其中一个构造器默认为EsAbortPolicy,它们还要求传入ThreadContext
  • 它覆盖了terminated、execute、afterExecute方法,其中terminated方法会回调listener.onTerminated();execute方法会捕获EsRejectedExecutionException异常,在command为AbstractRunnable类型时回调其onRejection及onAfter方法;afterExecute方法会执行EsExecutors.rethrowErrors(unwrap(r))方法
  • XRejectedExecutionHandler接口继承了RejectedExecutionHandler接口,它定义了rejected方法返回rejected的数量;它有两个实现类分别为EsAbortPolicy及ForceQueuePolicy
  • EsAbortPolicy实现了XRejectedExecutionHandler接口,其内部使用CounterMetric类维护rejected数量,而rejected方法直接返回该值;rejectedExecution方法对AbstractRunnable类型的runnable会判断是否isForceExecution,且是SizeBlockingQueue,则调用SizeBlockingQueue的forcePut方法重新force执行该runnable,之后就是递增rejected计数
  • ForceQueuePolicy实现了XRejectedExecutionHandler接口,它的rejectedExecution方法仅仅对ExecutorScalingQueue进行重新入队操作,而rejected方法返回0
  • AbstractRunnable声明实现Runnable接口,它的run方法分别会回调doRun、onFailure、onAfter方法;另外它还定义了isForceExecution方法用于确定当rejected的时候是否force execution

doc

© 著作权归作者所有

go4it
粉丝 89
博文 1102
码字总数 1039680
作品 0
深圳
私信 提问
聊聊Elasticsearch的SizeBlockingQueue

序 本文主要研究一下Elasticsearch的SizeBlockingQueue SizeBlockingQueue elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/SizeBlockingQueue.java Si......

go4it
06/01
17
0
聊聊Elasticsearch的FixedExecutorBuilder

序 本文主要研究一下Elasticsearch的FixedExecutorBuilder FixedExecutorBuilder elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/threadpool/FixedExecutorBuilder.java Fixed......

go4it
06/15
21
0
聊聊Elasticsearch的TimedRunnable

序 本文主要研究一下Elasticsearch的TimedRunnable TimedRunnable elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/TimedRunnable.java TimedRunnable继......

go4it
06/11
24
0
聊聊elasticsearch的SeedHostsResolver

序 本文主要研究一下elasticsearch的SeedHostsResolver ConfiguredHostsResolver elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java ConfiguredHostsR......

go4it
04/26
7
0
聊聊elasticsearch的ZenPing

序 本文主要研究一下elasticsearch的ZenPing ZenPing elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/zen/ZenPing.java ZenPing接口继承了Releasable接口,另外它还......

go4it
04/28
7
0

没有更多内容

加载失败,请刷新页面

加载更多

定期批量改密,实现高效运维,保障口令安全

随着企业IT资产规模的不断增大,各类主机、应用系统的管理也变得愈加困难。 对于系统管理员来说,保证操作系统的密码安全是其重要工作,在需要维护众多的主机时,其面临的困境将是: 1、难以...

堡垒啊
30分钟前
5
0
怎样在磁盘上查找MySQL表的大小?这里有答案

导读 我想知道 MySQL 表在磁盘上占用多少空间,但看起来很琐碎。不应该在 INFORMATION_SCHEMA.TABLES 中提供这些信息吗?没那么简单! 我想知道 MySQL 表在磁盘上占用多少空间,但看起来很琐碎...

问题终结者
56分钟前
6
0
jQuery load() 方法实现加载远程数据

jQuery load() 方法是简单但强大的 AJAX 方法。load() 方法从服务器加载数据,并把返回的数据放入被选元素中。 语法: $(selector).load(URL,data,callback);必需的 URL 参数规定您希望加载的...

前端老手
57分钟前
5
0
Spring Boot缓存实战 Redis 设置有效时间和自动刷新缓存-2

问题 上一篇Spring Boot Cache + redis 设置有效时间和自动刷新缓存,时间支持在配置文件中配置,说了一种时间方式,直接扩展注解的Value值,如: @Override@Cacheable(value = "people#${s...

xiaolyuh
今天
10
0
怎样在磁盘上查找MySQL表的大小?这里有答案

我想知道 MySQL 表在磁盘上占用多少空间,但看起来很琐碎。不应该在 INFORMATION_SCHEMA.TABLES 中提供这些信息吗?没那么简单! 我想知道 MySQL 表在磁盘上占用多少空间,但看起来很琐碎。不应...

Linux就该这么学
今天
5
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部