Java 线程池原理分析

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

+----ThreadPoolExecutor.Worker.java Worker(Runnable firstTask) { setState(-1); this.firstTask = firstTask; // 调用线程工厂创建线程 this.thread = getThreadFactory().newThread(this); } // Worker 实现了 Runnable 接口 public void run() { runWorker(this); } +----ThreadPoolExecutor.java final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); boolean completedAbruptly = true; try { // 循环从任务队列中获取新任务 while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { // 执行新任务 task.run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { // 线程退出后，进行后续处理 processWorkerExit(w, completedAbruptly); } }

+---- AbstractExecutorService.java public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); // 创建任务 RunnableFuture<Void> ftask = newTaskFor(task, null); // 提交任务 execute(ftask); return ftask; } +---- ThreadPoolExecutor.java public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // 如果工作线程数量 < 核心线程数，则创建新线程 if (workerCountOf(c) < corePoolSize) { // 添加工作者对象 if (addWorker(command, true)) return; c = ctl.get(); } // 缓存任务，如果队列已满，则 offer 方法返回 false。否则，offer 返回 true 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); } // 添加工作者对象，并在 addWorker 方法中检测线程数是否小于最大线程数 else if (!addWorker(command, false)) // 线程数 >= 最大线程数，使用拒绝策略处理任务 reject(command); } private boolean addWorker(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())) return false; for (;;) { int wc = workerCountOf(c); // 检测工作线程数与核心线程数或最大线程数的关系 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; 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 { int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); // 将 worker 对象添加到 workers 集合中 workers.add(w); int s = workers.size(); // 更新 largestPoolSize 属性 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 开始执行任务 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }