publicThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler){ ... }
int corePoolSize 核心线程数
int maximumPoolSize 非核心线程数
long keepAliveTime 运行线程等待任务时间
TimeUnit unit 时间单位
BlockingQueue workQueue 任务队列
ThreadFactory threadFactory 创建线程的工厂
RejectedExecutionHandler handler 拒绝策略
execute
1 2 3
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(4, 14, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(6), Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());
publicvoidexecute(Runnable command){ if (command == null) thrownew 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); elseif (workerCountOf(recheck) == 0) // 如果工作线程数为 0 ,( 因为有些线程可能已经消亡) // 补充工作线程 addWorker(null, false); } elseif (!addWorker(command, false)) reject(command); }
privatebooleanaddWorker(Runnable firstTask, boolean core){ // 先确保工作数量 添加上去了 retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 确认状态 // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) returnfalse; // 确认 工作线程数 添加成功 for (;;) { int wc = workerCountOf(c); if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) returnfalse; if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } // 在确保当前线程池状态, 已经 工作线程数的条件下 尝试添加工作线程 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable thrownew IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
privatevoidprocessWorkerExit(Worker w, boolean completedAbruptly){ if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount();
int c = ctl.get(); if (runStateLessThan(c, STOP)) { if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) min = 1; if (workerCountOf(c) >= min) return; // replacement not needed } addWorker(null, false); } }
