Java并发指南:ThreadPoolExecutor配置RejectedExecutionHandler指南及避坑

崔认知

发布于 2023-06-19 14:58:41

2360

发布于 2023-06-19 14:58:41

文章被收录于专栏：nobodynobody

ThreadPoolExecutor的RejectedExecutionHandler什么时候触发

当线程池关闭还在往线程池里添加任务
线程配置了有限的线程池大小及有限的工作队列，即线程池无法容纳新任务了

java为我们提供的RejectedExecutionHandler的几种实现及避坑

DiscardOldestPolicy

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardOldestPolicy} for the given executor.
         */
        public DiscardOldestPolicy() { }

        /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

丢弃队列头部即入队时间最早的任务，

提交当前任务。这种在业务场景下不

会采取，否则造成业务故障。

AbortPolicy

public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

直接抛出异常，当前任务不会提交到线程池队列中。由我们业务方自己处理当前不能处理的任务。这是线程池默认使用的拒绝任务的策略。

CallerRunsPolicy

public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code CallerRunsPolicy}.
         */
        public CallerRunsPolicy() { }

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

当前提交任务的线程执行任务。一般我们在把串行执行的业务逻辑拆分成并行，优化性能的时候使用（因为你也不知道线程池是否已满，退而求其次变成串行执行也无妨）。

DiscardPolicy

public static class DiscardPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardPolicy}.
         */
        public DiscardPolicy() { }

        /**
         * Does nothing, which has the effect of discarding task r.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

什么也不处理策略。这种策略我们坚决不使用，而且使用

java.util.concurrent.AbstractExecutorService#submit(java.util.concurrent.Callable<T>)

【避坑】：提交任务，使用java.util.concurrent.Future#get()会一直阻塞（强制使用java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit)）。

总结

java提供的RejectedExecutionHandler，我们一般不直接使用，而是选择继承方式，打印日志后再调用其默认实现。
DiscardPolicy在和java.util.concurrent.Future#get()一起使用时会一直阻塞（强制使用java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit)）。
CallerRunsPolicy在我们优化接口性能时，串行优化为并行选择使用（不要忘记监控线程池，调整线程池大小）。
DiscardOldestPolicy一般关键业务不要使用。
AbortPolicy需要业务方考虑当前不能提交的任务如何处理。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2022-04-09，如有侵权请联系 cloudcommunity@tencent.com 删除

java