这篇文章主要介绍了java 中ThreadPoolExecutor原理分析的相关资料,需要的朋友可以参考下
java 中ThreadPoolExecutor原理分析
线程池简介
Java线程池是开发中常用的工具，当我们有异步、并行的任务要处理时，经常会用到线程池，或者在实现一个服务器时，也需要使用线程池来接收连接处理请求。线程池使用JDK中提供的线程池实现位于java.util.concurrent.ThreadPoolExecutor。在使用时，通常使用ExecutorService接口，它提供了submit,invokeAll,shutdown等通用的方法。在线程池配置方面，Executors类中提供了一些静态方法能够提供一些常用场景的线程池，如newFixedThreadPool,newCachedThreadPool,newSingleThreadExecutor等，这些方法最终都是调用到了ThreadPoolExecutor的构造函数。ThreadPoolExecutor的包含所有参数的构造函数是/** * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @param threadFactory the factory to use when the executor * creates a new thread * @param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSizecorePoolSize设置线程池的核心线程数，当添加新任务时，如果线程池中的线程数小于corePoolSize，则不管当前是否有线程闲置，都会创建一个新的线程来执行任务。maximunPoolSize是线程池中允许的最大的线程数workQueue用于存放排队的任务keepAliveTime是大于corePoolSize的线程闲置的超时时间handler用于在任务逸出、线程池关闭时的任务处理 ，线程池的线程增长策略为，当前线程数小于corePoolSize时，新增线程，当线程数=corePoolSize且corePoolSize时，只有在workQueue不能存放新的任务时创建新线程，超出的线程在闲置keepAliveTime后销毁。实现(基于JDK1.8)ThreadPoolExecutor中保存的状态有当前线程池状态, 包括RUNNING,SHUTDOWN,STOP,TIDYING,TERMINATED。当前有效的运行线程的数量。将这两个状态放到一个int变量中，前三位作为线程池状态，后29位作为线程数量。例如0b11100000000000000000000000000001， 表示RUNNING, 一个线程。通过HashSet来存储工作者集合，访问该HashSet前必须先获取保护状态的mainLock:reentrantlocksubmit、executeexecute的执行方式为，首先检查当前worker数量，如果小于corePoolSize，则尝试add一个core Worker。线程池在维护线程数量以及状态检查上做了大量检测。public void execute(Runnable command) { int c = ctl.get(); // 如果当期数量小于corePoolSize if (workerCountOf(c)addWorker方法实现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 { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); if (rs largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 如果添加成功，则启动该线程，执行Worker的run方法，Worker的run方法执行外部的runWorker(Worker) t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }Worker类继承了AbstractQueuedSynchronizer获得了同步等待这样的功能。private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in. Null if factory fails. */ final Thread thread; /** Initial task to run. Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); } // Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state. protected boolean isHeldExclusively() { return getState() != 0; } protected boolean tryAcquire(int unused) { if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } protected boolean tryRelease(int unused) { setExclusiveOwnerThread(null); setState(0); return true; } public void lock() { acquire(1); } public boolean tryLock() { return tryAcquire(1); } public void unlock() { release(1); } public boolean isLocked() { return isHeldExclusively(); } void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } }runWorker(Worker)是Worker的轮询执行逻辑，不断地从工作队列中获取任务并执行它们。Worker每次执行任务前需要进行lock，防止在执行任务时被interrupt。final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts 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); } }ThreadPoolExecutor的submit方法中将Callable包装成FutureTask后交给execute方法。FutureTaskFutureTask继承于Runnable和Future，FutureTask定义的几个状态为NEW, 尚未执行COMPLETING, 正在执行norMAL, 正常执行完成得到结果EXCEPTIONAL, 执行抛出异常CANCELLED， 执行被取消INTERRUPTING，执行正在被中断INTERRUPTED, 已经中断。其中关键的get方法public V get() throws InterruptedException, ExecutionException { int s = state; if (s先获取当前状态，如果还未执行完成并且正常，则进入等待结果流程。在awaitDone不断循环获取当前状态，如果没有结果，则将自己通过CAS的方式添加到等待链表的头部，如果设置了超时，则LockSupport.parkNanos到指定的时间。static final class WaitNode { volatile Thread thread; volatile WaitNode next; WaitNode() { thread = Thread.currentThread(); } } private int awaitDone(boolean timed, long nanos) throws InterruptedException { final long deadline = timed ? System.nanoTime() + nanos : 0L; WaitNode q = null; boolean queued = false; for (;;) { if (Thread.interrupted()) { removeWaiter(q); throw new InterruptedException(); } int s = state; if (s > COMPLETING) { if (q != null) q.thread = null; return s; } else if (s == COMPLETING) // cannot time out yet Thread.yield(); else if (q == null) q = new WaitNode(); else if (!queued) queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q); else if (timed) { nanos = deadline - System.nanoTime(); if (nanosFutureTask的run方法是执行任务并设置结果的位置,首先判断当前状态是否为NEW并且将当前线程设置为执行线程，然后调用Callable的call获取结果后设置结果修改FutureTask状态。public void run() { if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { Callable c = callable; if (c != null && state == NEW) { V result; boolean ran; try { result = c.call(); ran = true; } catch (Throwable ex) { result = null; ran = false; setException(ex); } if (ran) set(result); } } finally { // runner must be non-null until state is settled to // prevent concurrent calls to run() runner = null; // state must be re-read after nulling runner to prevent // leaked interrupts int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！
