Java并发（五）线程池使用番外-分析RejectedExecutionException异常

时间 2019-12-11

标签 java 并发线程使用分析 rejectedexecutionexception 异常栏目 Java 繁體版

原文原文链接

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task org.cellphone.common.pool.Worker@f6f4d33 rejected from java.util.concurrent.ThreadPoolExecutor@23fc625e[Running, pool size = 3, active threads = 3, queued tasks = 15, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    at org.cellphone.common.pool.RejectedExecutionExceptionExample.main(RejectedExecutionExceptionExample.java:22)

1、入门示例

下面的测试程序使用 ThreadPoolExecutor 类来建立线程池执行任务，表明任务 Worker 类代码以下：测试

/**
 * Created by on 2019/4/20.
 */
public class Worker implements Runnable {

    private int id;

    public Worker(int id) {
        this.id = id;
    }

    @Override
    public void run() {
        try {
            System.out.println(Thread.currentThread().getName() + " 执行任务 " + id);
            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName() + " 完成任务 " + id);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

执行 Worker 任务的代码以下：this

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by on 2019/4/20.
 */
public class RejectedExecutionExceptionExample {

    public static void main(String[] args) {

        ExecutorService executor = new ThreadPoolExecutor(3, 3, 0L,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(15));

        Worker tasks[] = new Worker[10];
        for (int i = 0; i < 10; i++) {
            tasks[i] = new Worker(i);
            System.out.println("提交任务: " + tasks[i] + ", " + i);
            executor.execute(tasks[i]);
        }
        System.out.println("主线程结束");
        executor.shutdown();    // 关闭线程池
    }
}

运行一下，看到以下输出：spa

提交任务: org.cellphone.common.pool.Worker@36baf30c, 0
提交任务: org.cellphone.common.pool.Worker@5ca881b5, 1
提交任务: org.cellphone.common.pool.Worker@4517d9a3, 2
提交任务: org.cellphone.common.pool.Worker@2f92e0f4, 3
提交任务: org.cellphone.common.pool.Worker@28a418fc, 4
提交任务: org.cellphone.common.pool.Worker@5305068a, 5
提交任务: org.cellphone.common.pool.Worker@1f32e575, 6
提交任务: org.cellphone.common.pool.Worker@279f2327, 7
提交任务: org.cellphone.common.pool.Worker@2ff4acd0, 8
提交任务: org.cellphone.common.pool.Worker@54bedef2, 9
主线程结束
pool-1-thread-1 执行任务 0
pool-1-thread-2 执行任务 1
pool-1-thread-3 执行任务 2
pool-1-thread-1 完成任务 0
pool-1-thread-1 执行任务 3
pool-1-thread-2 完成任务 1
pool-1-thread-2 执行任务 4
pool-1-thread-3 完成任务 2
pool-1-thread-3 执行任务 5
pool-1-thread-2 完成任务 4
pool-1-thread-2 执行任务 6
pool-1-thread-3 完成任务 5
pool-1-thread-1 完成任务 3
pool-1-thread-3 执行任务 7
pool-1-thread-1 执行任务 8
pool-1-thread-3 完成任务 7
pool-1-thread-2 完成任务 6
pool-1-thread-1 完成任务 8
pool-1-thread-2 执行任务 9
pool-1-thread-2 完成任务 9

在 RejectedExecutionExceptionExample 类里，咱们使用 ThreadPoolExecutor 类建立了一个数量为3的线程池来执行任务，在这3个线程执行任务被占用期间，若是有新任务提交给线程池，那么这些新任务会被保存在 BlockingQueue 阻塞队列里，以等待被空闲线程取出并执行。在这里咱们使用一个大小为15的 ArrayBlockingQueue 队列来保存待执行的任务，而后咱们建立了10个任务提交给 ThreadPoolExecutor 线程池。线程

2、异常场景1

产生 RejectedExecutionException 异常的第一个缘由：code

调用 shutdown() 方法关闭了 ThreadPoolExecutor 线程池，又提交新任务给 ThreadPoolExecutor 线程池执行。通常调用 shutdown() 方法以后，JVM会获得一个关闭线程池的信号，并不会当即关闭线程池，原来线程池里未执行完的任务仍然在执行，等到任务都执行完后才关闭线程池，可是JVM不容许再提交新任务给线程池。

让咱们用如下例子来重现该异常：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by on 2019/4/20.
 */
public class RejectedExecutionExceptionExample {

    public static void main(String[] args) {

        ExecutorService executor = new ThreadPoolExecutor(3, 3, 0L,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(15));

        Worker tasks[] = new Worker[10];
        for (int i = 0; i < 10; i++) {
            tasks[i] = new Worker(i);
            System.out.println("提交任务: " + tasks[i] + ", " + i);
            executor.execute(tasks[i]);
        }
        System.out.println("主线程结束");
        executor.shutdown();// 关闭线程池
        executor.execute(tasks[0]);// 关闭线程池以后提交新任务，运行以后抛异常
    }
}

运行一下，看到以下输出：

提交任务: org.cellphone.common.pool.Worker@36baf30c, 0
提交任务: org.cellphone.common.pool.Worker@5ca881b5, 1
提交任务: org.cellphone.common.pool.Worker@4517d9a3, 2
提交任务: org.cellphone.common.pool.Worker@2f92e0f4, 3
提交任务: org.cellphone.common.pool.Worker@28a418fc, 4
提交任务: org.cellphone.common.pool.Worker@5305068a, 5
提交任务: org.cellphone.common.pool.Worker@1f32e575, 6
提交任务: org.cellphone.common.pool.Worker@279f2327, 7
提交任务: org.cellphone.common.pool.Worker@2ff4acd0, 8
提交任务: org.cellphone.common.pool.Worker@54bedef2, 9
主线程结束
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task org.cellphone.common.pool.Worker@36baf30c rejected from java.util.concurrent.ThreadPoolExecutor@5caf905d[Shutting down, pool size = 3, active threads = 3, queued tasks = 7, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    at org.cellphone.common.pool.RejectedExecutionExceptionExample.main(RejectedExecutionExceptionExample.java:26)
pool-1-thread-1 执行任务 0
pool-1-thread-2 执行任务 1
pool-1-thread-3 执行任务 2
pool-1-thread-1 完成任务 0
pool-1-thread-1 执行任务 3
pool-1-thread-2 完成任务 1
pool-1-thread-2 执行任务 4
pool-1-thread-3 完成任务 2
pool-1-thread-3 执行任务 5
pool-1-thread-3 完成任务 5
pool-1-thread-3 执行任务 6
pool-1-thread-2 完成任务 4
pool-1-thread-2 执行任务 7
pool-1-thread-1 完成任务 3
pool-1-thread-1 执行任务 8
pool-1-thread-3 完成任务 6
pool-1-thread-2 完成任务 7
pool-1-thread-3 执行任务 9
pool-1-thread-1 完成任务 8
pool-1-thread-3 完成任务 9

从以上例子能够看出，在调用 shutdown() 方法以后，因为JVM不容许再提交新任务给线程池，因而抛出了 RejectedExecutionException 异常。

3、异常场景2

产生 RejectedExecutionException 异常第二个缘由：

要提交给阻塞队列的任务超出了该队列的最大容量。当线程池里的线程都繁忙的时候，新任务会被提交给阻塞队列保存，这个阻塞队列一旦饱和，线程池就会拒绝接收新任务，随即抛出异常。

示例代码以下：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by on 2019/4/20.
 */
public class RejectedExecutionExceptionExample {

    public static void main(String[] args) {

        ExecutorService executor = new ThreadPoolExecutor(3, 3, 0L,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(15));

        // 提交20个任务给线程池
        Worker tasks[] = new Worker[20];
        for (int i = 0; i < 20; i++) {
            tasks[i] = new Worker(i);
            System.out.println("提交任务: " + tasks[i] + ", " + i);
            executor.execute(tasks[i]);
        }
        System.out.println("主线程结束");
        executor.shutdown();// 关闭线程池
    }
}

在上面的例子中，咱们使用了一个大小为15的 ArrayBlockingQueue 阻塞队列来保存等待执行的任务。接着咱们提交了20个任务给线程池，因为每一个线程执行任务的时候会睡眠1秒，所以当3个线程繁忙的时候，其余任务不会当即获得执行，咱们提交的新任务会被保存在队列里。当等待任务的数量超过线程池阻塞队列的最大容量时，抛出了 RejectedExecutionException 异常。

4、解决方法

要解决 RejectedExecutionException 异常，首先咱们要注意两种状况：

当调用了线程池的shutdown()方法之后，不要提交新任务给线程池
不要提交大量超过线程池处理能力的任务，这时可能会致使队列饱和，抛出异常

对于第二种状况，咱们很容易解决。咱们能够选择一种不须要设置大小限制的数据结构，好比 LinkedBlockingQueue 阻塞队列。所以在使用 LinkedBlockingQueue 队列之后，若是还出现 RejectedExecutionException 异常，就要将问题的重点放在第一种状况上。若是第一种状况不是产生问题的缘由，那么咱们还须要寻找更复杂的缘由。好比，因为线程死锁和 LinkedBlockingQueue 饱和，致使内存占用过大，这个时候咱们就须要考虑JVM可用内存的问题了。

对于第二种状况，一般有一些隐藏的信息被咱们忽略。其实咱们能够给使用 ArrayBlockingQueue 做为阻塞队列的 ThreadPoolExecutor 线程池提交超过15个的任务，只要咱们在提交新任务前设置一个完成原来任务的等待时间，这时3个线程就会逐渐消费 ArrayBlockingQueue 阻塞队列里的任务，而不会使它堵塞。示例以下：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by on 2019/4/20.
 */
public class RejectedExecutionExceptionExample {

    public static void main(String[] args) throws InterruptedException {

        ExecutorService executor = new ThreadPoolExecutor(3, 3, 0L,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(15));

        // 提交20个任务给线程池
        Worker tasks[] = new Worker[20];
        for (int i = 0; i < 10; i++) {
            tasks[i] = new Worker(i);
            System.out.println("提交任务: " + tasks[i] + ", " + i);
            executor.execute(tasks[i]);
        }

        Thread.sleep(3000);// 让主线程睡眠三秒
        for (int i = 10; i < 20; i++) {
            tasks[i] = new Worker(i);
            System.out.println("提交任务: " + tasks[i] + ", " + i);
            executor.execute(tasks[i]);
        }

        System.out.println("主线程结束");
        executor.shutdown();// 关闭线程池
    }
}