Java线程池拒绝策略

现在线程池有一个任务队列，用于缓存所有待处理的任务，正在处理的任务将从任务队列中移除。因此在任务队列长度有限的情况下,再添加任务就会出现任务被拒绝加入到队列处理的情况，需要有一种策略来处理应该加入任务队列却因为队列已满无法加入的情况。另外在线程池关闭的时候也需要对任务加入队列操作进行额外的协调处理。

在JDK的RejectedExecutionHandler接口中,提供了四种方式来处理任务拒绝策略:

1、直接丢弃（DiscardPolicy）

代码

package com.courage;


import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/*
直接丢弃策略
*/

public class DiscardPolicyDemo {
    private static SimpleDateFormat sdf  = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    public static void main(String[] args) {
        //定义线程池核心数
        int corePoolSize = 1;
        //定义线程池最大连接
        int maximumPoolSize = 1;
        //阻塞队列
        BlockingQueue queue = new ArrayBlockingQueue<Runnable>(1);
        //创建线程池
        ThreadPoolExecutor pool = new ThreadPoolExecutor(corePoolSize,  maximumPoolSize,
                0, TimeUnit.SECONDS, queue ) ;
        //设置线程池拒绝策略
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy ());
        //不断往线程池提交任务
        for(int i=0;i<10;i++){
            final int index = i;
            pool.submit(new Runnable(){
                @Override
                public void run() {
                    log(Thread.currentThread().getName()+"begin run task :"+index);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log(Thread.currentThread().getName()+" finish run  task :"+index);
                }
            });
        }
        log("main thread before sleep!!!");
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log("before shutdown()");

        pool.shutdown();

        log("after shutdown(),pool.isTerminated=" + pool.isTerminated());
        try {
            pool.awaitTermination(1000L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log("now,pool.isTerminated=" + pool.isTerminated());
    }
    //重写log方法
    protected static void log(String string) {
        System.out.println(sdf.format(new Date())+"  "+string);
    }
}

运行结果

2020-12-24 16:23:45  main thread before sleep!!!
2020-12-24 16:23:45  pool-1-thread-2begin run task :2
2020-12-24 16:23:45  pool-1-thread-1begin run task :0
2020-12-24 16:23:45  pool-1-thread-3begin run task :3
2020-12-24 16:23:46  pool-1-thread-3 finish run  task :3
2020-12-24 16:23:46  pool-1-thread-1 finish run  task :0
2020-12-24 16:23:46  pool-1-thread-2 finish run  task :2
2020-12-24 16:23:46  pool-1-thread-3begin run task :1
2020-12-24 16:23:47  pool-1-thread-3 finish run  task :1
2020-12-24 16:23:49  before shutdown()
2020-12-24 16:23:49  after shutdown(),pool.isTerminated=false
2020-12-24 16:23:49  now,pool.isTerminated=true

只有task0、task1、task2、task3四个任务被执行了,其余的六个任务被丢弃

为什么只有四个任务被执行了呢？

过程是这样的：由于我们的任务队列的容量为3.当task0正在执行的时候，task1、task2、task3被提交到了队列中但是还没有执行，受队列容量的限制，submit提交的task4~task9就都被直接抛弃了。因此就只有task0、task1、task2、task3被执行了。

2、丢弃队列中最老的任务(DiscardOldestPolicy)

代码

package com.courage;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class DiscardOldestPolicyDemo {
    private static SimpleDateFormat sdf  = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    public static void main(String[] args) {
        //定义线程池核心数
        int corePoolSize = 1;
        //定义线程池最大连接
        int maximumPoolSize = 1;
        //阻塞队列
        BlockingQueue queue = new ArrayBlockingQueue<Runnable>(1);
        //创建线程池
        ThreadPoolExecutor pool = new ThreadPoolExecutor(corePoolSize,  maximumPoolSize,
                0, TimeUnit.SECONDS, queue ) ;
        //设置线程池拒绝策略
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        //不断往线程池提交任务
        for(int i=0;i<10;i++){
            final int index = i;
            pool.submit(new Runnable(){
                @Override
                public void run() {
                    log(Thread.currentThread().getName()+"begin run task :"+index);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log(Thread.currentThread().getName()+" finish run  task :"+index);
                }
            });
        }
        log("main thread before sleep!!!");
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log("before shutdown()");

        pool.shutdown();

        log("after shutdown(),pool.isTerminated=" + pool.isTerminated());
        try {
            pool.awaitTermination(1000L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log("now,pool.isTerminated=" + pool.isTerminated());
    }
    //重写log方法
    protected static void log(String string) {
        System.out.println(sdf.format(new Date())+"  "+string);
    }
}

运行结果

2020-12-24 19:10:54  main thread before sleep!!!
2020-12-24 19:10:54  pool-1-thread-1begin run task :0
2020-12-24 19:10:55  pool-1-thread-1 finish run  task :0
2020-12-24 19:10:55  pool-1-thread-1begin run task :9
2020-12-24 19:10:56  pool-1-thread-1 finish run  task :9
2020-12-24 19:10:58  before shutdown()
2020-12-24 19:10:58  after shutdown(),pool.isTerminated=false
2020-12-24 19:10:58  now,pool.isTerminated=true

为什么只有task0和task9执行了呢?

因为开始执行任务task0时,这时task1进入队列中,task2进入队列中会顶替掉task1,task3会顶替掉task2,直到task9顶替掉8,只有task9在任务队列中,最后被执行。

3、抛异常(AbortPolicy)

代码同上

只修改策略为AbortPolicy

运行结果

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@433c675d[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@2e817b38[Wrapped task = com.courage.DiscardOldestPolicyDemo$1@c4437c4]] rejected from java.util.concurrent.ThreadPoolExecutor@3f91beef[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0]
    at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055)
    at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
    at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355)
    at java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:118)
    at com.courage.DiscardOldestPolicyDemo.main(DiscardOldestPolicyDemo.java:28)
2020-12-24 19:15:35  pool-1-thread-1begin run task :0
2020-12-24 19:15:36  pool-1-thread-1 finish run  task :0
2020-12-24 19:15:36  pool-1-thread-1begin run task :1
2020-12-24 19:15:37  pool-1-thread-1 finish run  task :1

当task0开始执行时,task1进入队列,之后的任务抛出异常。

4、分给调用线程来执行(CallerRunsPolicy)

代码同上

只修改策略为CallerRunsPolicy

运行结果

2020-12-24 19:19:33  pool-1-thread-1begin run task :0
2020-12-24 19:19:33  mainbegin run task :2
2020-12-24 19:19:34  main finish run  task :2
2020-12-24 19:19:34  pool-1-thread-1 finish run  task :0
2020-12-24 19:19:34  mainbegin run task :3
2020-12-24 19:19:34  pool-1-thread-1begin run task :1
2020-12-24 19:19:35  main finish run  task :3
2020-12-24 19:19:35  pool-1-thread-1 finish run  task :1
2020-12-24 19:19:35  pool-1-thread-1begin run task :4
2020-12-24 19:19:35  mainbegin run task :5
2020-12-24 19:19:36  main finish run  task :5
2020-12-24 19:19:36  pool-1-thread-1 finish run  task :4
2020-12-24 19:19:36  pool-1-thread-1begin run task :6
2020-12-24 19:19:36  mainbegin run task :8
2020-12-24 19:19:37  main finish run  task :8
2020-12-24 19:19:37  pool-1-thread-1 finish run  task :6
2020-12-24 19:19:37  pool-1-thread-1begin run task :7
2020-12-24 19:19:37  main thread before sleep!!!
2020-12-24 19:19:38  pool-1-thread-1 finish run  task :7
2020-12-24 19:19:38  pool-1-thread-1begin run task :9
2020-12-24 19:19:39  pool-1-thread-1 finish run  task :9
2020-12-24 19:19:41  before shutdown()
2020-12-24 19:19:41  after shutdown(),pool.isTerminated=false
2020-12-24 19:19:41  now,pool.isTerminated=true

从结果可以看出，没有任务被抛弃，而是将由的任务分配到main线程中执行了。

小结

这四种策略是独立无关的，是对任务拒绝处理的四中表现形式。最简单的方式就是直接丢弃任务。但是却有两种方式，到底是该丢弃哪一个任务，比如可以丢弃当前将要加入队列的任务本身（DiscardPolicy）或者丢弃任务队列中最旧任务（DiscardOldestPolicy）。丢弃最旧任务也不是简单的丢弃最旧的任务，而是有一些额外的处理。除了丢弃任务还可以直接抛出一个异常（RejectedExecutionException），这是比较简单的方式。抛出异常的方式（AbortPolicy）尽管实现方式比较简单，但是由于抛出一个RuntimeException，因此会中断调用者的处理过程。除了抛出异常以外还可以不进入线程池执行，在这种方式（CallerRunsPolicy）中任务将有调用者线程去执行。