/** * 一种ExecutorService * 它使用几个池线程类型中的一个执行每个提交的任务 * 通常使用Executors工厂方法配置创建 * * An {@link ExecutorService} that executes each submitted task using * one of possibly several pooled threads, normally configured * using {@link Executors} factory methods. * * 线程池解决了两个不同的问题: * 1.它们通常在执行大量异步任务时提供更好的性能,这是由于减少了每个任务的调用开销; * 2.并且它们提供了一种限制和管理资源(包括执行任务集合时消耗的线程)的方法 * * 每个ThreadPoolExecutor还维护一些基本统计信息,例如已完成的任务数 * * <p>Thread pools address two different problems: they usually * provide improved performance when executing large numbers of * asynchronous tasks, due to reduced per-task invocation overhead, * and they provide a means of bounding and managing the resources, * including threads, consumed when executing a collection of tasks. * Each {@code ThreadPoolExecutor} also maintains some basic * statistics, such as the number of completed tasks. * * 为了在各种上下文中(场景)使用 * 这个类提供了许多可调整的参数和可扩展的hook方法 * 但是,程序员应该使用更方便的Executors工厂方法 * 1.Executors{newCachedThreadPool}(无限制线程池,自动回收线程) * 2.Executors{newFixedThreadPool}(固定大小的线程池) * 3.Executors{newSingleThreadExecutor}(单后台线程) * 为最常见的使用场景预先配置设置 * * 否则,在手动配置和调整此类时,请使用以下指南: * * <p>To be useful across a wide range of contexts, this class * provides many adjustable parameters and extensibility * hooks. However, programmers are urged to use the more convenient * {@link Executors} factory methods {@link * Executors#newCachedThreadPool} (unbounded thread pool, with * automatic thread reclamation), {@link Executors#newFixedThreadPool} * (fixed size thread pool) and {@link * Executors#newSingleThreadExecutor} (single background thread), that * preconfigure settings for the most common usage * scenarios. Otherwise, use the following guide when manually * configuring and tuning this class: * * <dl> * * 核心池和最大池大小 * * <dt>Core and maximum pool sizes</dt> * * ThreadPoolExecutor将根据: * corePoolSize和maximumPoolSize设置的界限自动调整池大小 * * <dd>A {@code ThreadPoolExecutor} will automatically adjust the * pool size (see {@link #getPoolSize}) * according to the bounds set by * corePoolSize (see {@link #getCorePoolSize}) and * maximumPoolSize (see {@link #getMaximumPoolSize}). * * 在方法execute(Runnable)中提交新任务时 * 如果正在运行的线程少于corePoolSize,则会创建一个新线程来处理该请求 * (即使其他工作线程处于空闲状态) * * 否则,如果运行的线程少于maximumPoolSize,则仅当队列已满时,才会创建一个新线程来处理该请求 * 1.通过将corePoolSize和maximumPoolSize设置为相同的值,可以创建一个固定大小的线程池 * 2.通过将maximumPoolSize设置为本质上无边界的值,例如Integer.MAX_value,可以允许池容纳任意数量的并发任务 * 最典型的情况是,核心池和最大池大小仅在构造时设置 * 但也可以使用setCorePoolSize和setMaximumPoolSize动态更改它们 * * When a new task is submitted in method {@link #execute(Runnable)}, * if fewer than corePoolSize threads are running, a new thread is * created to handle the request, even if other worker threads are * idle. Else if fewer than maximumPoolSize threads are running, a * new thread will be created to handle the request only if the queue * is full. By setting corePoolSize and maximumPoolSize the same, you * create a fixed-size thread pool. By setting maximumPoolSize to an * essentially unbounded value such as {@code Integer.MAX_VALUE}, you * allow the pool to accommodate an arbitrary number of concurrent * tasks. Most typically, core and maximum pool sizes are set only * upon construction, but they may also be changed dynamically using * {@link #setCorePoolSize} and {@link #setMaximumPoolSize}. </dd> * * 按需构造 * * <dt>On-demand construction</dt> * * 默认情况下,即使核心线程最初也是在新任务到达时创建和启动的 * 也可以使用方法prestartCoreThread或prestartcorethreads动态重写 * 如果使用非空队列构造线程池,则可能需要预启动线程 * * <dd>By default, even core threads are initially created and * started only when new tasks arrive, but this can be overridden * dynamically using method {@link #prestartCoreThread} or {@link * #prestartAllCoreThreads}. You probably want to prestart threads if * you construct the pool with a non-empty queue. </dd> * * 创建新线程 * * <dt>Creating new threads</dt> * * 使用ThreadFactory创建新线程 * 如果未另行指定,则使用Executors.defaultThreadFactory * 它创建的线程都处于相同的ThreadGroup,并且具有相同的NORM_PRIORITY优先级和非守护进程状态 * 通过提供不同的ThreadFactory,可以更改线程的名称、线程组、优先级、守护进程状态等 * * 如果ThreadFactory在从newThread返回null时无法创建线程,则执行器将继续,但可能无法执行任何任务 * 线程应该拥有modifyThread(RuntimePermission) * 如果工作线程或使用池的其他线程不具有此权限,则服务可能会降级: * 配置更改可能不会及时生效,关闭池可能仍处于可以终止但未完成的状态 * * <dd>New threads are created using a {@link ThreadFactory}. If not * otherwise specified, a {@link Executors#defaultThreadFactory} is * used, that creates threads to all be in the same {@link * ThreadGroup} and with the same {@code NORM_PRIORITY} priority and * non-daemon status. By supplying a different ThreadFactory, you can * alter the thread's name, thread group, priority, daemon status, * etc. If a {@code ThreadFactory} fails to create a thread when asked * by returning null from {@code newThread}, the executor will * continue, but might not be able to execute any tasks. Threads * should possess the "modifyThread" {@code RuntimePermission}. If * worker threads or other threads using the pool do not possess this * permission, service may be degraded: configuration changes may not * take effect in a timely manner, and a shutdown pool may remain in a * state in which termination is possible but not completed.</dd> * * 线程生存时间 * * <dt>Keep-alive times</dt> * * 如果池当前有多个corePoolSize线程,则如果多余的线程空闲时间超过keepAliveTime,则将终止这些线程 * 这提供了一种在池未被积极使用时减少资源消耗的方法 * 如果线程池稍后变得更为活跃,则将构造新线程 * 也可以使用方法setKeepAliveTime(long,TimeUnit)动态更改此参数 * * 使用Long.MAX_value, TimeUnit.NANOSECONDS的值可以有效地禁止空闲线程在关闭之前终止 * 默认情况下,keep-alive策略仅在存在多个corePoolSize线程时应用 * 但是只要keep alive time值非零,也可以使用方法allowCoreThreadTimeOut(true) * 将此超时策略应用于核心线程 * * <dd>If the pool currently has more than corePoolSize threads, * excess threads will be terminated if they have been idle for more * than the keepAliveTime (see {@link #getKeepAliveTime(TimeUnit)}). * This provides a means of reducing resource consumption when the * pool is not being actively used. If the pool becomes more active * later, new threads will be constructed. This parameter can also be * changed dynamically using method {@link #setKeepAliveTime(long, * TimeUnit)}. Using a value of {@code Long.MAX_VALUE} {@link * TimeUnit#NANOSECONDS} effectively disables idle threads from ever * terminating prior to shut down. By default, the keep-alive policy * applies only when there are more than corePoolSize threads, but * method {@link #allowCoreThreadTimeOut(boolean)} can be used to * apply this time-out policy to core threads as well, so long as the * keepAliveTime value is non-zero. </dd> * * 队列 * * <dt>Queuing</dt> * * 任何BlockingQueue都可以用来传输和保存提交的任务, 此队列的使用与池大小有关: * * <dd>Any {@link BlockingQueue} may be used to transfer and hold * submitted tasks. The use of this queue interacts with pool sizing: * * <ul> * * 如果运行的线程少于corePoolSize,则执行器总是倾向于添加新线程,而不是排队 * * <li>If fewer than corePoolSize threads are running, the Executor * always prefers adding a new thread * rather than queuing. * * 如果corePoolSize或更多线程正在运行,则执行器总是希望将请求排队,而不是添加新线程 * * <li>If corePoolSize or more threads are running, the Executor * always prefers queuing a request rather than adding a new * thread. * * 如果请求不能排队,则创建一个新线程 * 除非该线程超过maximumPoolSize,在这种情况下,任务将被拒绝 * * <li>If a request cannot be queued, a new thread is created unless * this would exceed maximumPoolSize, in which case, the task will be * rejected. * * </ul> * * 排队的一般策略有三种: * * There are three general strategies for queuing: * <ol> * * 直接交接 * 工作队列的一个很好的默认选择是SynchronousQueue,它将任务直接交给线程,而不必保存它们 * 在这里,如果没有线程可以立即运行任务,则尝试对任务进行排队将失败,因此将构造一个新线程 * 此策略在处理可能具有内部依赖项的请求集时避免锁定 * 直接切换通常需要无限的队列大小,以避免拒绝新提交的任务 * 这反过来又造成了当命令的平均到达速度超过可以处理的速度时,无限线程增长的可能性 * * <li><em> Direct handoffs.</em> A good default choice for a work * queue is a {@link SynchronousQueue} that hands off tasks to threads * without otherwise holding them. Here, an attempt to queue a task * will fail if no threads are immediately available to run it, so a * new thread will be constructed. This policy avoids lockups when * handling sets of requests that might have internal dependencies. * Direct handoffs generally require unbounded maximumPoolSizes to * avoid rejection of new submitted tasks. This in turn admits the * possibility of unbounded thread growth when commands continue to * arrive on average faster than they can be processed. * * 无界队列 * 使用无界队列(例如LinkedBlockingQueue没有预定义的容量) * 将导致新任务在所有corePoolSize线程忙时在队列中等待 * 因此,不会创建超过corePoolSize的线程(所以maximumPoolSize的值没有任何影响) * 当每个任务完全独立于其他任务时,任务不会影响彼此的执行; * 例如,在web页面服务器中 * 虽然这种排队方式有助于消除请求的暂时性突发 * 但它允许在命令平均到达速度超过其处理速度时,无限工作队列增长的可能性 * * <li><em> Unbounded queues.</em> Using an unbounded queue (for * example a {@link LinkedBlockingQueue} without a predefined * capacity) will cause new tasks to wait in the queue when all * corePoolSize threads are busy. Thus, no more than corePoolSize * threads will ever be created. (And the value of the maximumPoolSize * therefore doesn't have any effect.) This may be appropriate when * each task is completely independent of others, so tasks cannot * affect each others execution; for example, in a web page server. * While this style of queuing can be useful in smoothing out * transient bursts of requests, it admits the possibility of * unbounded work queue growth when commands continue to arrive on * average faster than they can be processed. * * 有界队列 * 有界队列(例如ArrayBlockingQueue) * 有助于防止在与有限的maximumPoolSizes一起使用时资源耗尽,但可能更难进行优化和控制 * 队列大小和最大池大小可以相互交换: * 1.使用大队列和小池可以最小化CPU使用率、操作系统资源和上下文切换开销,但可能导致人为的低吞吐量; 如果任务经常阻塞(例如,如果它们是I/O绑定的),系统可能(比创建更多的线程运行时)使用更长时间; * 2.使用小队列通常需要较大的池大小,这使得cpu更加繁忙,可能会遇到不可接受的调度开销,这也会降低吞吐量 * * <li><em>Bounded queues.</em> A bounded queue (for example, an * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when * used with finite maximumPoolSizes, but can be more difficult to * tune and control. Queue sizes and maximum pool sizes may be traded * off for each other: Using large queues and small pools minimizes * CPU usage, OS resources, and context-switching overhead, but can * lead to artificially low throughput. If tasks frequently block (for * example if they are I/O bound), a system may be able to schedule * time for more threads than you otherwise allow. Use of small queues * generally requires larger pool sizes, which keeps CPUs busier but * may encounter unacceptable scheduling overhead, which also * decreases throughput. * * </ol> * * </dd> * * 拒绝策略 * * <dt>Rejected tasks</dt> * * 在方法execute(Runnable)中提交的新任务将在: * 1.执行器关闭 * 2.执行器对最大线程工作队列容量使用有限边界且已饱和 * 时被拒绝 * 在这两种情况下,execute方法都会调用其 * RejectedExecutionHandler的RejectedExecutionHandler方法的RejectedExecutionHandler} * * JDK提供了四个预定义的处理程序策略: * * <dd>New tasks submitted in method {@link #execute(Runnable)} will be * <em>rejected</em> when the Executor has been shut down, and also when * the Executor uses finite bounds for both maximum threads and work queue * capacity, and is saturated. In either case, the {@code execute} method * invokes the {@link * RejectedExecutionHandler#rejectedExecution(Runnable, ThreadPoolExecutor)} * method of its {@link RejectedExecutionHandler}. Four predefined handler * policies are provided: * * <ol> * * 默认的ThreadPoolExecutor.AbortPolicy: * 处理程序在拒绝时抛出运行时RejectedExecutionException * * <li>In the default {@link ThreadPoolExecutor.AbortPolicy}, the handler * throws a runtime {@link RejectedExecutionException} upon rejection. * * ThreadPoolExecutor.CallerRunsPolicy: * 调用execute本身的线程运行该任务 * 这提供了一个简单的反馈控制机制,可以降低提交新任务的速度 * * <li>In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread * that invokes {@code execute} itself runs the task. This provides a * simple feedback control mechanism that will slow down the rate that * new tasks are submitted. * * ThreadPoolExecutor.DiscardPolicy: * 无法执行的任务将被简单地删除 * * <li>In {@link ThreadPoolExecutor.DiscardPolicy}, a task that * cannot be executed is simply dropped. * * ThreadPoolExecutor.DiscardOldestPolicy: * 如果未关闭执行器,则删除工作队列头部的任务,然后重试执行 * (这可能会再次失败,导致重复执行) * * <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the * executor is not shut down, the task at the head of the work queue * is dropped, and then execution is retried (which can fail again, * causing this to be repeated.) * * </ol> * * 可以定义和使用其他类型的RejectedExecutionHandler类 * 这样做需要注意,特别是当策略设计为仅在特定容量或排队策略下工作时 * * It is possible to define and use other kinds of {@link * RejectedExecutionHandler} classes. Doing so requires some care * especially when policies are designed to work only under particular * capacity or queuing policies. </dd> * * 钩子方法 * * <dt>Hook methods</dt> * * 此类提供在执行每个任务之前和之后调用的(protected)可重写的beforeExecute和afterExecute方法 * 它们可用于操作执行环境, 例如: * 重新初始化ThreadLocals、收集统计信息或添加日志项 * 此外,可以重写方法terminated,以便在执行器完全终止后执行任何需要执行的特殊处理 * * <dd>This class provides {@code protected} overridable * {@link #beforeExecute(Thread, Runnable)} and * {@link #afterExecute(Runnable, Throwable)} methods that are called * before and after execution of each task. These can be used to * manipulate the execution environment; for example, reinitializing * ThreadLocals, gathering statistics, or adding log entries. * Additionally, method {@link #terminated} can be overridden to perform * any special processing that needs to be done once the Executor has * fully terminated. * * 如果hook、callback或BlockingQueue方法抛出异常 * 则内部工作线程可能会依次失败、突然终止并可能被替换 * * <p>If hook, callback, or BlockingQueue methods throw exceptions, * internal worker threads may in turn fail, abruptly terminate, and * possibly be replaced.</dd> * * 队列维护 * * <dt>Queue maintenance</dt> * * 方法getQueue()允许访问工作队列以进行监视和调试 * 强烈建议不要将此方法用于任何其他目的 * 提供的两个方法remove和purge可用于在取消大量排队任务时协助存储回收 * * <dd>Method {@link #getQueue()} allows access to the work queue * for purposes of monitoring and debugging. Use of this method for * any other purpose is strongly discouraged. Two supplied methods, * {@link #remove(Runnable)} and {@link #purge} are available to * assist in storage reclamation when large numbers of queued tasks * become cancelled.</dd> * * 回收 * * <dt>Reclamation</dt> * * 程序中不再引用且没有剩余线程的池可以在不显式关闭的情况下回收(垃圾回收) * 通过设置适当的保持活动时间、使用零核心线程的下限和/或设置allowCoreThreadTimeOut(true) * 可以将池配置为允许所有未使用的线程最终消亡 * * <dd>A pool that is no longer referenced in a program <em>AND</em> * has no remaining threads may be reclaimed (garbage collected) * without being explicitly shutdown. You can configure a pool to * allow all unused threads to eventually die by setting appropriate * keep-alive times, using a lower bound of zero core threads and/or * setting {@link #allowCoreThreadTimeOut(boolean)}. </dd> * * </dl> * * 扩展例子: * 这个类的大多数扩展重写了一个或多个受保护的钩子方法 * 例如,下面是一个子类,它添加了一个简单的暂停/恢复功能: * * <p><b>Extension example</b>. Most extensions of this class * override one or more of the protected hook methods. For example, * here is a subclass that adds a simple pause/resume feature: * * <pre> {@code * class PausableThreadPoolExecutor extends ThreadPoolExecutor { * private boolean isPaused; * private ReentrantLock pauseLock = new ReentrantLock(); * private Condition unpaused = pauseLock.newCondition(); * * public PausableThreadPoolExecutor(...) { super(...); } * // 重写执行前方法 * protected void beforeExecute(Thread t, Runnable r) { * super.beforeExecute(t, r); * pauseLock.lock(); * try { * while (isPaused) unpaused.await(); * } catch (InterruptedException ie) { * t.interrupt(); * } finally { * pauseLock.unlock(); * } * } * // 停止 * public void pause() { * pauseLock.lock(); * try { * isPaused = true; * } finally { * pauseLock.unlock(); * } * } * // 恢复 * public void resume() { * pauseLock.lock(); * try { * isPaused = false; * unpaused.signalAll(); * } finally { * pauseLock.unlock(); * } * } * }}</pre> * * @since 1.5 * @author Doug Lea */ publicclassThreadPoolExecutorextendsAbstractExecutorService { /** * 主池控制状态ctl属性是一个原子整数, 包含两个概念字段: * 1.workerCount: 指示线程的有效运行状态数; * 2.runState: 指示是否正在运行、正在关闭等; * * The main pool control state, ctl, is an atomic integer packing * two conceptual fields * workerCount, indicating the effective number of threads * runState, indicating whether running, shutting down etc * * 为了将它们打包成一个int * 我们将workerCount限制为(2^29)-1[大约5亿]线程,而不是(2^31)-1[20亿]线程 * 如果将来出现这种情况(线程数超过5亿): * 可以将变量更改为AtomicLong, 并调整下面的shift/mask常量 * 但是在需要这样改变之前,使用int代码会更快更简单 * * In order to pack them into one int, we limit workerCount to * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2 * billion) otherwise representable. If this is ever an issue in * the future, the variable can be changed to be an AtomicLong, * and the shift/mask constants below adjusted. But until the need * arises, this code is a bit faster and simpler using an int. * * workerCount是允许启动和不允许停止的工人数 * 该值可能暂时不同于活动线程的实际数量, 例如: * 1.当线程工厂在被请求时无法创建线程 * 2.当退出的线程在终止前仍在执行记帐(仍然在数量上, 未被减一)时 * 用户可见池大小为工作集的当前大小 * * The workerCount is the number of workers that have been * permitted to start and not permitted to stop. The value may be * transiently different from the actual number of live threads, * for example when a ThreadFactory fails to create a thread when * asked, and when exiting threads are still performing * bookkeeping before terminating. The user-visible pool size is * reported as the current size of the workers set. * * 运行状态提供主生命周期控制,具有以下值: * * RUNNING: 接受新任务并处理排队的任务 * SHUTDOWN: 不接受新任务,但处理排队的任务 * STOP: 不接受新任务、不处理排队的任务和中断正在进行的任务 * TIDYING: 所有任务都已终止,workerCount为零, 状态转换为tiding的线程将运行terminated()钩子方法 * TERMINATED: terminated()方法已完成 * * The runState provides the main lifecycle control, taking on values: * * RUNNING: Accept new tasks and process queued tasks * SHUTDOWN: Don't accept new tasks, but process queued tasks * STOP: Don't accept new tasks, don't process queued tasks, * and interrupt in-progress tasks * TIDYING: All tasks have terminated, workerCount is zero, * the thread transitioning to state TIDYING * will run the terminated() hook method * TERMINATED: terminated() has completed * * 这些值之间的数字顺序很重要,以便进行有序比较 * 运行状态随时间单调地增加,但不必命中每个状态, 状态过渡是: * * RUNNING -> SHUTDOWN: 调用shutdown()方法时 * * (RUNNING or SHUTDOWN) -> STOP: 在调用shutdownNow()方法时 * * SHUTDOWN -> TIDYING: 当队列和池都为空时 * * STOP -> TIDYING: 当池为空时 * * TIDYING -> TERMINATED: 当terminated()钩子方法完成时 * * The numerical order among these values matters, to allow * ordered comparisons. The runState monotonically increases over * time, but need not hit each state. The transitions are: * * RUNNING -> SHUTDOWN * On invocation of shutdown() * (RUNNING or SHUTDOWN) -> STOP * On invocation of shutdownNow() * SHUTDOWN -> TIDYING * When both queue and pool are empty * STOP -> TIDYING * When pool is empty * TIDYING -> TERMINATED * When the terminated() hook method has completed * * 等待awaitTermination()的线程将在状态达到TERMINATED时返回 * * Threads waiting in awaitTermination() will return when the * state reaches TERMINATED. * * 检测从SHUTDOWN到TIDYING的转换比想象的要简单得多 * 因为队列在(正常工作)非空之后可能变空,在SHUTDOWN状态下也可能变空 * 但我们只在看到它为空之后又看到workerCount为0(有时需要重新检查——见下文)时终止 * * Detecting the transition from SHUTDOWN to TIDYING is less * straightforward than you'd like because the queue may become * empty after non-empty and vice versa during SHUTDOWN state, but * we can only terminate if, after seeing that it is empty, we see * that workerCount is 0 (which sometimes entails a recheck -- see * below). */ privatefinalAtomicIntegerctl=newAtomicInteger(ctlOf(RUNNING, 0)); privatestaticfinalintCOUNT_BITS= Integer.SIZE - 3; privatestaticfinalintCOUNT_MASK= (1 << COUNT_BITS) - 1;
// TODO: switch to AbstractQueuedLongSynchronizer and move // completedTasks into the lock word.
/** * 使用给定的第一个任务和ThreadFactory创建线程 * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ Worker(Runnable firstTask) { setState(-1); // 禁止中断runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); }
// Check if queue empty only if necessary. if (runStateAtLeast(c, SHUTDOWN) && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) { decrementWorkerCount(); returnnull; }
publicThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
publicThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler); }
publicThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler); }
/** * 设置用于创建新线程的线程工厂 * * @param threadFactory the new thread factory * @throws NullPointerException if threadFactory is null * @see #getThreadFactory */ publicvoidsetThreadFactory(ThreadFactory threadFactory) { if (threadFactory == null) thrownewNullPointerException(); this.threadFactory = threadFactory; }
/** * Returns the thread factory used to create new threads. * * @return the current thread factory * @see #setThreadFactory(ThreadFactory) */ publicThreadFactorygetThreadFactory() { return threadFactory; }
/** * 为不可执行的任务设置新的处理程序(拒绝策略) * * @param handler the new handler * @throws NullPointerException if handler is null * @see #getRejectedExecutionHandler */ publicvoidsetRejectedExecutionHandler(RejectedExecutionHandler handler) { if (handler == null) thrownewNullPointerException(); this.handler = handler; }
/** * Returns the current handler for unexecutable tasks. * * @return the current handler * @see #setRejectedExecutionHandler(RejectedExecutionHandler) */ publicRejectedExecutionHandlergetRejectedExecutionHandler() { return handler; }
/** * 设置线程池的核心数, 这将重写构造函数中设置的值 * 如果新值小于当前值,则多余的现有线程将在下次变为空闲时而终止 * 如果更大,如果有新任务需要,将启动新线程来执行任何排队的任务 * * @param corePoolSize the new core size * @throws IllegalArgumentException if {@code corePoolSize < 0} * or {@code corePoolSize} is greater than the {@linkplain * #getMaximumPoolSize() maximum pool size} * @see #getCorePoolSize */ publicvoidsetCorePoolSize(int corePoolSize) { // 如果corePoolSize<0或者corePoolSize大于线程池最大值 if (corePoolSize < 0 || maximumPoolSize < corePoolSize) thrownewIllegalArgumentException(); // 增量delta intdelta= corePoolSize - this.corePoolSize; // 更新corePoolSize this.corePoolSize = corePoolSize; // 现有的工人数大于corePoolSize(即数量减少) if (workerCountOf(ctl.get()) > corePoolSize) interruptIdleWorkers(); // 现有的工人数小于corePoolSize(数量要增加) elseif (delta > 0) { // 我们真的不知道“需要”多少个新线程 // 作为一种启发式方法,预先启动足够多的新工作线程(最多新的核心大小) // 来处理队列中当前数量的任务,但在执行此操作时,如果队列变为空,则停止 intk= Math.min(delta, workQueue.size()); while (k-- > 0 && addWorker(null, true)) { if (workQueue.isEmpty()) break; } } }
/** * 返回线程池的核心数 * * @return the core number of threads * @see #setCorePoolSize */ publicintgetCorePoolSize() { return corePoolSize; }
/** * 提前启动"一个"核心线程,让它空闲地等待工作 * 这将覆盖仅在执行新任务时启动核心线程的默认策略 * 如果所有核心线程都已启动,则此方法将返回false * * @return {@code true} if a thread was started */ publicbooleanprestartCoreThread() { return workerCountOf(ctl.get()) < corePoolSize && addWorker(null, true); }
/** * 提前启动所有核心线程,让它们空闲地等待工作 * 这将覆盖仅在执行新任务时启动核心线程的默认策略 * * @return the number of threads started */ publicintprestartAllCoreThreads() { intn=0; while (addWorker(null, true)) ++n; return n; }
/** * 如果此池允许核心线程超时,则返回true; * 如果在keepAlive时间内没有任务到达,线程terminate; * 如果需要,则在新任务到达时替换; * 如果为true,则应用于非核心线程的相同保持活动策略也适用于核心线程 * 如果为false(默认值),则不会由于缺少传入任务而终止核心线程 * * @return {@code true} if core threads are allowed to time out, * else {@code false} * * @since 1.6 */ publicbooleanallowsCoreThreadTimeOut() { return allowCoreThreadTimeOut; }
/** * 设置控制"核心线程"是否可以超时的策略 * 如果在保持活动时间内没有任务到达,则该策略将在新任务到达时根据需要被替换 * 如果为false,则不会由于缺少传入任务而终止核心线程; * 如果为true,则应用于非核心线程的相同保持活动策略也适用于核心线程; * 为了避免连续的线程替换,在设置true时keepAliveTime必须大于零 * 通常应该在"线程池被激活之前调用此方法" * * @param value {@code true} if should time out, else {@code false} * @throws IllegalArgumentException if value is {@code true} * and the current keep-alive time is not greater than zero * * @since 1.6 */ publicvoidallowCoreThreadTimeOut(boolean value) { // 超时时间应当大于零 if (value && keepAliveTime <= 0) thrownewIllegalArgumentException("Core threads must have nonzero keep alive times"); // 更新超时状态标志 if (value != allowCoreThreadTimeOut) { allowCoreThreadTimeOut = value; if (value) interruptIdleWorkers(); } }
/** * 设置允许的最大线程数, 这将重写构造函数中设置的值 * 如果新值小于当前值,则多余的现有线程将在下次变为空闲时终止 * * @param maximumPoolSize the new maximum * @throws IllegalArgumentException if the new maximum is * less than or equal to zero, or * less than the {@linkplain #getCorePoolSize core pool size} * @see #getMaximumPoolSize */ publicvoidsetMaximumPoolSize(int maximumPoolSize) { if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize) thrownewIllegalArgumentException(); this.maximumPoolSize = maximumPoolSize; // 如果新值小于当前值,则检测一次空闲线程 if (workerCountOf(ctl.get()) > maximumPoolSize) interruptIdleWorkers(); }
/** * Returns the maximum allowed number of threads. * * @return the maximum allowed number of threads * @see #setMaximumPoolSize */ publicintgetMaximumPoolSize() { return maximumPoolSize; }
/** * 设置线程keepAliveTime,即线程在终止之前可以保持空闲的时间量 * 如果池中当前的线程数超过核心线程数; * 或者如果此池allowsCoreThreadTimeOut允许核心线程超时, * 则到此时间而空闲的线程将被终止 * 这将重写构造函数中设置的值 * * @param time 时间值为零将导致"多余的线程"在执行任务后"立即终止" * 允许allowsCoreThreadTimeOut时, 不可设置为0 * @param unit the time unit of the {@code time} argument * @throws IllegalArgumentException if {@code time} less than zero or * if {@code time} is zero and {@code allowsCoreThreadTimeOut} * @see #getKeepAliveTime(TimeUnit) */ publicvoidsetKeepAliveTime(long time, TimeUnit unit) { if (time < 0) thrownewIllegalArgumentException(); // 允许allowsCoreThreadTimeOut时, 不可设置为0 if (time == 0 && allowsCoreThreadTimeOut()) thrownewIllegalArgumentException("Core threads must have nonzero keep alive times"); // 设置时间 longkeepAliveTime= unit.toNanos(time); longdelta= keepAliveTime - this.keepAliveTime; this.keepAliveTime = keepAliveTime; // 如果等待时间减少, 清除一次空闲线程 if (delta < 0) interruptIdleWorkers(); }
/** * 返回线程保持活动时间,即线程在终止之前可能保持空闲的时间量 * * @param unit the desired time unit of the result * @return the time limit * @see #setKeepAliveTime(long, TimeUnit) */ publiclonggetKeepAliveTime(TimeUnit unit) { return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS); }
/** * 返回正在积极执行任务的线程的大概数量(根据worker被锁来判断) * * @return the number of threads */ publicintgetActiveCount() { finalReentrantLockmainLock=this.mainLock; mainLock.lock(); try { intn=0; for (Worker w : workers) // 根据worker被锁来判断 if (w.isLocked()) ++n; return n; } finally { mainLock.unlock(); } }
/** * 返回池中同时存在的最大线程数 * * @return the number of threads */ publicintgetLargestPoolSize() { finalReentrantLockmainLock=this.mainLock; mainLock.lock(); try { return largestPoolSize; } finally { mainLock.unlock(); } }
/** * 返回计划执行和已执行的任务的大致总数 * 由于任务和线程的状态在计算期间可能会动态变化,因此返回的值只是一个近似值 * * @return the number of tasks */ publiclonggetTaskCount() { finalReentrantLockmainLock=this.mainLock; mainLock.lock(); try { // 当前统计完成的任务数 longn= completedTaskCount; // 遍历的原因是只有worker被清除时才将他的数据加入completedTaskCount属性中; for (Worker w : workers) { // 与当前工作线程的任务数相加 n += w.completedTasks; // 如果当前线程被锁, 则再加一 if (w.isLocked()) ++n; } // 最后加上任务队列长度(计划执行任务数) return n + workQueue.size(); } finally { mainLock.unlock(); } }
/** * 返回已完成执行的任务的大致总数 * 由于任务和线程的状态在计算过程中可能会动态变化,因此返回的值只是一个近似值 * 但在连续的调用中此值一定不会减少 * * @return the number of tasks */ publiclonggetCompletedTaskCount() { finalReentrantLockmainLock=this.mainLock; mainLock.lock(); try { longn= completedTaskCount; for (Worker w : workers) n += w.completedTasks; return n; } finally { mainLock.unlock(); } }
/** * Returns a string identifying this pool, as well as its state, * including indications of run state and estimated worker and * task counts. * * @return a string identifying this pool, as well as its state */ publicStringtoString() { long ncompleted; int nworkers, nactive; finalReentrantLockmainLock=this.mainLock; mainLock.lock(); try { ncompleted = completedTaskCount; nactive = 0; nworkers = workers.size(); for (Worker w : workers) { ncompleted += w.completedTasks; if (w.isLocked()) ++nactive; } } finally { mainLock.unlock(); } intc= ctl.get(); StringrunState= isRunning(c) ? "Running" : runStateAtLeast(c, TERMINATED) ? "Terminated" : "Shutting down"; returnsuper.toString() + "[" + runState + ", pool size = " + nworkers + ", active threads = " + nactive + ", queued tasks = " + workQueue.size() + ", completed tasks = " + ncompleted + "]"; }
/* 扩展的钩子方法 */
/** * 在给定线程中执行给定Runnable之前调用的方法 * 此方法由执行任务r的线程t调用,可用于重新初始化线程局部变量或执行日志记录 * * 这个实现什么也不做,但是可以"在子类中定制" * 注意:为了正确地嵌套多个重写,子类通常应该在这个"方法的末尾"调用super.beforeExecute * * @param t the thread that will run task {@code r} * @param r the task that will be executed */ protectedvoidbeforeExecute(Thread t, Runnable r) { }
/** * 在执行完给定的可运行程序后调用的方法, 此方法由执行任务的线程调用 * 如果异常非空,则抛出的是导致执行线程突然终止的未捕获的RuntimeException或Error * * 这个实现什么也不做,但是可以在子类中定制 * 注意: 为了正确地嵌套多个重写, 子类通常应该在"这个方法的开头"调用super.afterExecute * * 注意: 当操作显式地或通过方法(如submit)包含在任务(如FutureTask)中时: * 由这些任务对象捕获并维护计算异常,因此它们不会导致突然终止,并且内部异常不会传递给此方法; * * 如果你希望在此方法中捕获这两种类型的异常,则可以参考下面这个例子; * * 在此示例子类中,如果任务已中止,则打印直接原因或基础异常: * * <pre> {@code * class ExtendedExecutor extends ThreadPoolExecutor { * // ... * protected void afterExecute(Runnable r, Throwable t) { * super.afterExecute(r, t); * if (t == null * && r instanceof Future<?> * && ((Future<?>)r).isDone()) { * try { * Object result = ((Future<?>) r).get(); * } catch (CancellationException ce) { * t = ce; * } catch (ExecutionException ee) { * t = ee.getCause(); * } catch (InterruptedException ie) { * // ignore/reset * Thread.currentThread().interrupt(); * } * } * if (t != null) * System.out.println(t); * } * }}</pre> * * @param r the runnable that has completed * @param t the exception that caused termination, or null if * execution completed normally */ protectedvoidafterExecute(Runnable r, Throwable t) { }
/** * 在调用方线程中执行任务r,除非已关闭执行器,在这种情况下,将丢弃该任务 * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ publicvoidrejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { r.run(); } } }
/** * 不执行任何操作,直接丢弃任务r * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ publicvoidrejectedExecution(Runnable r, ThreadPoolExecutor e) { } }
/** * 拒绝策略: DiscardOldestPolicy * 丢弃最旧的未处理请求,然后重试execute,除非该执行程序关闭,否则该任务将被丢弃 */ publicstaticclassDiscardOldestPolicyimplementsRejectedExecutionHandler { /** * Creates a {@code DiscardOldestPolicy} for the given executor. */ publicDiscardOldestPolicy() { }
/** * 获取并忽略执行器将要执行的下一个任务(如果该任务立即可用) * 然后重试执行任务r,除非执行器关闭,否则将放弃任务r * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ publicvoidrejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { e.getQueue().poll(); e.execute(r); } } } }
呼呼~ 终于翻译完了2100多行的程序, 下面来做一下总结吧!
一.ctl属性
在ThreadLocalExecutor源码中第一个属性便是一个AtomicInteger类型的ctl, 用来描述线程池的状态(pool control state)