當往一個固定隊列ArrayBlockingQueue 不停的提交任務時，會發生什么？
請看如下代碼
private static final int QUEUE_SIZE = 20;
private static final int CORE_POOL_SIZE = 2;
private static final int MAX_POOL_SIZE = 2;
private static final int KEEP_ALIVE_TIME = 5;
public static void main(String[] args) throws Exception {
ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(QUEUE_SIZE));
while(true) {
SenderTask st = new SenderTask();
executor.submit(st);
}
}
run完后會發生如下異常：
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@b99f7c6 rejected from java.util.concurrent.ThreadPoolExecutor@2959ee1d[Running, pool size = 2, active threads = 2, queued tasks = 20, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2048)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:110)
如果你有研究過executor的源碼你將發現，每次在submit任務的時候，會先進行addWorker()的判斷，如果不能添加成功，則會拋出RejectedExecutionException
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
以上這個異常還有一種出現的情況就是，當你執行完execute.shutdown()后，任然往executor里提交task,也會拋出該異常
示例代碼如下：
public class RejectedExecutionExceptionExample {
public static void main(String[] args) {
ExecutorService executor = new ThreadPoolExecutor(3, 3, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(15));
Worker tasks[] = new Worker[20];
for(int i=0; i<20; i++){
tasks[i] = new Worker(i);
executor.execute(tasks[i]);
}
executor.shutdown();
executor.execute(tasks[0]);//繼續執行任務
}
}
如何解決呢？
1. 控制提交的任務數量，即提交的任務數量不要超過它當前能處理的能力 (這里可以用生產者消費者的模式來解決)
2. 確保不要在shutdown()之后在執行任務
3. 可以用LinkedBlockingQueue代替ArrayBlockingQueue,因為LinkedBlockingQueue可以設成無界的，但是需要注意，設成無界后最終可能發生OOM（內存溢出），
所以要保證第一二點。
參考文獻：https://examples.javacodegeeks.com/core-java/util/concurrent/rejectedexecutionexception/java-util-concurrent-rejectedexecutionexception-how-to-solve-rejectedexecutionexception/