bilibili-Java并发学习笔记19 线程池 ThreadPoolExecutor
基于 java 1.8.0
Executor
ExecutorService
AbstractExecutorService
ThreadPoolExecutor -> Worker
创建 ThreadPoolExecutor 实例
package new_package.thread.p52; import java.util.concurrent.*; public class ThreadPoolTest { public static void main(String[] args) throws ExecutionException, InterruptedException { ExecutorService executorService = new ThreadPoolExecutor(10, 10, 1, TimeUnit.SECONDS, new LinkedBlockingQueue(), (r, executor) -> { }); executorService.execute(() -> { System.out.println("ThreadPool"); }); Future<String> future = executorService.submit(() -> { int i = 77; return "hello " + i; }); System.out.println(future.get()); executorService.shutdown(); } }Executors 工厂模式 创建线程池
package new_package.thread.p52; import java.util.concurrent.*; public class ThreadPoolTest2 { public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(10); for (int i = 0; i < 1000; i++) { executorService.execute(() -> System.out.println(Thread.currentThread().getName())); } executorService.shutdown(); } }ThreadPoolExecutor 的构造参数
corePoolSize核心线程数
maximumPoolSize最大线程数
keepAliveTime
TimeUnit
与 keepAliveTime 结合使用 ， 当 maximumPoolSize > corePoolSize 时才有意义 当线程池的线程数量 > corePoolSize 时，且当前任务数并没有占满所有线程池中的线程，等到 keepAliveTime 后，线程将被回收；
BlockingQueue 阻塞队列 ArrayBlockingQueue 有界队列基于数组 LinkedBlockingQueue 有界队列基于链表吞吐量比 ArrayBlockingQueue 高 PriorityBlockingQueueDelayQueueSynchronousQueueLinkedTransferQueue ThreadFactory 线程工厂创建新线程并交由线程池管理，默认为 Executors.defaultThreadFactory()
public static ThreadFactory defaultThreadFactory() { return new DefaultThreadFactory(); } /** * Executors * The default thread factory */ static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } public Thread newThread(Runnable r) { Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); // 5 -> Normal priority for a thread return t; } }无法执行且无法存储的线程就进入拒绝策略
package java.util.concurrent; /** * 无法由 ThreadPoolExecutor 执行的任务的处理程序。 * * @since 1.5 * @author Doug Lea */ public interface RejectedExecutionHandler { /** * 当 execute 不能接受某个任务时，可以由 ThreadPoolExecutor 调用的方法。 * 因为超出其界限而没有更多可用的线程或队列槽时，或者关闭 Executor 时就可能发生这种情况。 * * 在没有其他替代方法的情况下，该方法可能抛出未经检查的 RejectedExecutionException， * 而该异常将传播到 execute 的调用者。 * * @param r 所请求执行的可运行任务。 * @param executor 试图执行此任务的执行程序。 * @throws RejectedExecutionException 如果没有补救方法。 */ void rejectedExecution(Runnable r, ThreadPoolExecutor executor); } /** * 默认拒绝策略 */ private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();ThreadPoolExecutor 提供的拒绝策略
/** * 不使用线程池中的线程执行，而是在当前线程中直接执行 */ public static class CallerRunsPolicy implements RejectedExecutionHandler { /** * Creates a {@code CallerRunsPolicy}. */ public CallerRunsPolicy() { } /** * Executes task r in the caller's thread, unless the executor * has been shut down, in which case the task is discarded. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { r.run(); } } } /** * 抛出异常 */ public static class AbortPolicy implements RejectedExecutionHandler { public AbortPolicy() { } /** * 总是抛出 RejectedExecutionException. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task * @throws RejectedExecutionException is RuntimeException */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString()); } } /** * 将任务直接丢弃，什么也不做 */ public static class DiscardPolicy implements RejectedExecutionHandler { public DiscardPolicy() { } /** * Does nothing, which has the effect of discarding task r. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { } } /** * 将阻塞队列中的队首的任务丢弃，将当前任务执行 execute 方法 */ public static class DiscardOldestPolicy implements RejectedExecutionHandler { public DiscardOldestPolicy() { } /** * Obtains and ignores the next task that the executor * would otherwise execute, if one is immediately available, * and then retries execution of task r, unless the executor * is shut down, in which case task r is instead discarded. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { e.getQueue().poll(); e.execute(r); } } }
线程池中有这样两个状态属性：线程池状态、线程池中线程数量；
线程池设计者使用一个字段 ctl 保存这两个状态属性
/** * 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 * * 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. * * 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 为零，转换为状态整理的线程将运行 terminated() 钩子方法 * TERMINATED: terminated() 方法执行完成 * * 为了进行有序比较，这些值之间的数字顺序很重要。运行状态随时间单调地增加，但不必触及每个状态。过渡包括： * * RUNNING -> SHUTDOWN * 调用 shutdown(), 可能隐含在 finalize() 中 * (RUNNING or SHUTDOWN) -> STOP * 调用 shutdownNow() * SHUTDOWN -> TIDYING * 当队列和线程池数量都为空时 * STOP -> TIDYING * 当线程池数量为空时 * TIDYING -> TERMINATED * When the terminated() hook method has completed * * 在 awaitermination() 中等待的线程将在状态达到 TERMINATED 时返回。 * * 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). */ // 初始值 ： 11100000 00000000 00000000 00000000 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // Integer.SIZE = 32 // COUNT_BITS = 29 private static final int COUNT_BITS = Integer.SIZE - 3; // 1 << COUNT_BITS --> 00100000 00000000 00000000 00000000 // after -1 --> 00011111 11111111 11111111 11111111 // 即 int 后 29 位用于存储线程池数量 // 即线程池最大数量为 2^29-1 ==> 5.36亿 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits // int 前 3 位用于存储 线程池数量 // -1 ==> 11111111 11111111 11111111 11111111 // 0 右移 29 位 = 11100000 00000000 00000000 00000000 private static final int RUNNING = -1 << COUNT_BITS; // 0 右移 29 位 = 00000000 00000000 00000000 00000000 private static final int SHUTDOWN = 0 << COUNT_BITS; // 1 右移 29 位 = 00100000 00000000 00000000 00000000 private static final int STOP = 1 << COUNT_BITS; // 2 ==> 10 // 2 右移 29 位 = 01000000 00000000 00000000 00000000 private static final int TIDYING = 2 << COUNT_BITS; // 3 ==> 011 // 3 右移 29 位 = 01100000 00000000 00000000 00000000 private static final int TERMINATED = 3 << COUNT_BITS; // Packing and unpacking ctl // 线程池状态 // c & 11100000 00000000 00000000 00000000 private static int runStateOf(int c) { return c & ~CAPACITY; } // 线程池数量 // c & 00011111 11111111 11111111 11111111 private static int workerCountOf(int c) { return c & CAPACITY; } // 逻辑或 private static int ctlOf(int rs, int wc) { return rs | wc; }ThreadPoolExecutor 源码
/** * 在将来某个时间执行给定任务。可以在新线程中或者在现有池线程中执行该任务。 * * 如果无法将任务提交执行，或者因为此执行程序已关闭，或者因为已达到其容量， * 则该任务由当前 RejectedExecutionHandler 处理。 * * @param command 要执行的任务。 * @throws RejectedExecutionException 如果无法接收要执行的任务， * 则由 RejectedExecutionHandler 决定是否抛出 RejectedExecutionException * @throws NullPointerException 如果 command 为 null */ 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(); // 当前线程池数量 < corePoolSize if (workerCountOf(c) < corePoolSize) { // 添加一个 Worker 线程执行任务 if (addWorker(command, true)) return; // 执行任务失败(多线程提交任务)，重新获取 ctl c = ctl.get(); } // 线程池状态为 RUNNING // 任务放入任务队列成功 if (isRunning(c) && workQueue.offer(command)) { // 重新获取 ctl int recheck = ctl.get(); // 如果线程池状态不是运行状态，则将 command 从任务队列中移除（回滚） if (! isRunning(recheck) && remove(command)) // 并将本任务进入拒绝策略 reject(command); // 线程池数量 = 0 ？？？ else if (workerCountOf(recheck) == 0) // addWorker(null, false); } // 线程池 else if (!addWorker(command, false)) reject(command); }