聊聊dubbo的EagerThreadPool-白红宇

序

本文主要研究一下dubbo的EagerThreadPool

EagerThreadPool

dubbo-2.7.2/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/eager/EagerThreadPool.java

public class EagerThreadPool implements ThreadPool {    @Override    public Executor getExecutor(URL url) {        String name = url.getParameter(THREAD_NAME_KEY, DEFAULT_THREAD_NAME);        int cores = url.getParameter(CORE_THREADS_KEY, DEFAULT_CORE_THREADS);        int threads = url.getParameter(THREADS_KEY, Integer.MAX_VALUE);        int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);        int alive = url.getParameter(ALIVE_KEY, DEFAULT_ALIVE);        // init queue and executor        TaskQueue
    
      taskQueue = new TaskQueue
     
      (queues <= 0 ? 1 : queues);        EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,                threads,                alive,                TimeUnit.MILLISECONDS,                taskQueue,                new NamedInternalThreadFactory(name, true),                new AbortPolicyWithReport(name, url));        taskQueue.setExecutor(executor);        return executor;    }}复制代码

EagerThreadPool实现了ThreadPool接口，其getExecutor创建的是EagerThreadPoolExecutor，它使用的queue为TaskQueue，使用的threadFactory为NamedInternalThreadFactory，使用的rejectedExecutionHandler为AbortPolicyWithReport

EagerThreadPoolExecutor

dubbo-2.7.2/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/eager/EagerThreadPoolExecutor.java

public class EagerThreadPoolExecutor extends ThreadPoolExecutor {    /**     * task count     */    private final AtomicInteger submittedTaskCount = new AtomicInteger(0);    public EagerThreadPoolExecutor(int corePoolSize,                                   int maximumPoolSize,                                   long keepAliveTime,                                   TimeUnit unit, TaskQueue
    
      workQueue,                                   ThreadFactory threadFactory,                                   RejectedExecutionHandler handler) {        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);    }    /**     * @return current tasks which are executed     */    public int getSubmittedTaskCount() {        return submittedTaskCount.get();    }    @Override    protected void afterExecute(Runnable r, Throwable t) {        submittedTaskCount.decrementAndGet();    }    @Override    public void execute(Runnable command) {        if (command == null) {            throw new NullPointerException();        }        // do not increment in method beforeExecute!        submittedTaskCount.incrementAndGet();        try {            super.execute(command);        } catch (RejectedExecutionException rx) {            // retry to offer the task into queue.            final TaskQueue queue = (TaskQueue) super.getQueue();            try {                if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {                    submittedTaskCount.decrementAndGet();                    throw new RejectedExecutionException("Queue capacity is full.", rx);                }            } catch (InterruptedException x) {                submittedTaskCount.decrementAndGet();                throw new RejectedExecutionException(x);            }        } catch (Throwable t) {            // decrease any way            submittedTaskCount.decrementAndGet();            throw t;        }    }}复制代码

EagerThreadPoolExecutor继承了ThreadPoolExecutor，它维护了submittedTaskCount，在执行任务之前递增，在afterExecute的时候胡递减；其execute方法会捕获RejectedExecutionException，然后使用TaskQueue的retryOffer再重新入队，入队不成功才抛出RejectedExecutionException

TaskQueue

dubbo-2.7.2/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/eager/TaskQueue.java

public class TaskQueue
    
      extends LinkedBlockingQueue
     
       {    private static final long serialVersionUID = -2635853580887179627L;    private EagerThreadPoolExecutor executor;    public TaskQueue(int capacity) {        super(capacity);    }    public void setExecutor(EagerThreadPoolExecutor exec) {        executor = exec;    }    @Override    public boolean offer(Runnable runnable) {        if (executor == null) {            throw new RejectedExecutionException("The task queue does not have executor!");        }        int currentPoolThreadSize = executor.getPoolSize();        // have free worker. put task into queue to let the worker deal with task.        if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {            return super.offer(runnable);        }        // return false to let executor create new worker.        if (currentPoolThreadSize < executor.getMaximumPoolSize()) {            return false;        }        // currentPoolThreadSize >= max        return super.offer(runnable);    }    /**     * retry offer task     *     * @param o task     * @return offer success or not     * @throws RejectedExecutionException if executor is terminated.     */    public boolean retryOffer(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {        if (executor.isShutdown()) {            throw new RejectedExecutionException("Executor is shutdown!");        }        return super.offer(o, timeout, unit);    }}复制代码

TaskQueue继承了LinkedBlockingQueue，它覆盖了offer方法，该方法在submittedTaskCount小于poolSize的时候会入队，如果大于等于poolSize则再判断currentPoolThreadSize是否小于maximumPoolSize，如果小于则返回false让线程池创建新线程，最后在currentPoolThreadSize大于等于maximumPoolSize的时候入队

NamedInternalThreadFactory

dubbo-2.7.2/dubbo-common/src/main/java/org/apache/dubbo/common/threadlocal/NamedInternalThreadFactory.java

public class NamedInternalThreadFactory extends NamedThreadFactory {    public NamedInternalThreadFactory() {        super();    }    public NamedInternalThreadFactory(String prefix) {        super(prefix, false);    }    public NamedInternalThreadFactory(String prefix, boolean daemon) {        super(prefix, daemon);    }    @Override    public Thread newThread(Runnable runnable) {        String name = mPrefix + mThreadNum.getAndIncrement();        InternalThread ret = new InternalThread(mGroup, runnable, name, 0);        ret.setDaemon(mDaemon);        return ret;    }}复制代码

NamedInternalThreadFactory继承了NamedThreadFactory，这里创建的是InternalThread

AbortPolicyWithReport

dubbo-2.7.2/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/AbortPolicyWithReport.java

public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {    protected static final Logger logger = LoggerFactory.getLogger(AbortPolicyWithReport.class);    private final String threadName;    private final URL url;    private static volatile long lastPrintTime = 0;    private static final long TEN_MINUTES_MILLS = 10 * 60 * 1000;    private static final String OS_WIN_PREFIX = "win";    private static final String OS_NAME_KEY = "os.name";    private static final String WIN_DATETIME_FORMAT = "yyyy-MM-dd_HH-mm-ss";    private static final String DEFAULT_DATETIME_FORMAT = "yyyy-MM-dd_HH:mm:ss";    private static Semaphore guard = new Semaphore(1);    public AbortPolicyWithReport(String threadName, URL url) {        this.threadName = threadName;        this.url = url;    }    @Override    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {        String msg = String.format("Thread pool is EXHAUSTED!" +                " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: "                + "%d)," +                " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",            threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(),            e.getLargestPoolSize(),            e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),            url.getProtocol(), url.getIp(), url.getPort());        logger.warn(msg);        dumpJStack();        throw new RejectedExecutionException(msg);    }    private void dumpJStack() {        long now = System.currentTimeMillis();        //dump every 10 minutes        if (now - lastPrintTime < TEN_MINUTES_MILLS) {            return;        }        if (!guard.tryAcquire()) {            return;        }        ExecutorService pool = Executors.newSingleThreadExecutor();        pool.execute(() -> {            String dumpPath = url.getParameter(DUMP_DIRECTORY, System.getProperty("user.home"));            SimpleDateFormat sdf;            String os = System.getProperty(OS_NAME_KEY).toLowerCase();            // window system don't support ":" in file name            if (os.contains(OS_WIN_PREFIX)) {                sdf = new SimpleDateFormat(WIN_DATETIME_FORMAT);            } else {                sdf = new SimpleDateFormat(DEFAULT_DATETIME_FORMAT);            }            String dateStr = sdf.format(new Date());            //try-with-resources            try (FileOutputStream jStackStream = new FileOutputStream(                new File(dumpPath, "Dubbo_JStack.log" + "." + dateStr))) {                JVMUtil.jstack(jStackStream);            } catch (Throwable t) {                logger.error("dump jStack error", t);            } finally {                guard.release();            }            lastPrintTime = System.currentTimeMillis();        });        //must shutdown thread pool ,if not will lead to OOM        pool.shutdown();    }}复制代码

AbortPolicyWithReport继承了ThreadPoolExecutor.AbortPolicy，其rejectedExecution方法会输出包含thread pool相关信息的msg，然后使用warn级别打印出来，然后进行dumpJStack，最后再抛出RejectedExecutionException

小结

EagerThreadPool实现了ThreadPool接口，其getExecutor创建的是EagerThreadPoolExecutor，它使用的queue为TaskQueue，使用的threadFactory为NamedInternalThreadFactory，使用的rejectedExecutionHandler为AbortPolicyWithReport

序

EagerThreadPool

EagerThreadPoolExecutor

TaskQueue

NamedInternalThreadFactory

AbortPolicyWithReport

小结

doc