怎么制作私人网站教程,vs 网站开发,大沥南海网站建设,制作外贸网站模板下载Java中的线程池
谈到池化#xff0c;通常我们会想到线程池、连接池、对象池等池子。通常池化都会有以下这些共同的好处#xff1a;
减少资源的消耗。在线程池中重复利用已经创建的线程可以减少创建和销毁的损耗。方便资源的管理。在线程池中我们可以制定线程数量、什么情况…Java中的线程池
谈到池化通常我们会想到线程池、连接池、对象池等池子。通常池化都会有以下这些共同的好处
减少资源的消耗。在线程池中重复利用已经创建的线程可以减少创建和销毁的损耗。方便资源的管理。在线程池中我们可以制定线程数量、什么情况下会创建线程、线程满了怎么拒绝等。
在Java线程池通常指的是ThreadPoolExecutor。
类的声明
public class ThreadPoolExecutor extends AbstractExecutorService {}构造方法
ThreadPoolExecutor提供以下几个构造方法
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue)public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory)public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, RejectedExecutionHandler handler)public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
实际上所有的构造方法都是最后一个构造方法的重载:
当没有提供线程工厂ThreadFactory时使用Executors.defaultThreadFactory。当没有提供拒绝策略RejectedExecutionHandler时使用defaultHandler, 这个defaultHandler对应的是抛弃策略AbortPolicy。 // 抛弃策略private static final RejectedExecutionHandler defaultHandler new AbortPolicy();public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler);}public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, defaultHandler);}public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, RejectedExecutionHandler handler) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), handler);}public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {if (corePoolSize 0 ||maximumPoolSize 0 ||maximumPoolSize corePoolSize ||keepAliveTime 0)throw new IllegalArgumentException();if (workQueue null || threadFactory null || handler null)throw new NullPointerException();this.corePoolSize corePoolSize;this.maximumPoolSize maximumPoolSize;this.workQueue workQueue;this.keepAliveTime unit.toNanos(keepAliveTime);this.threadFactory threadFactory;this.handler handler;}这里虽然提供了很多的参数但是需要了解这些参数的作用需要往下看ThreadPoolExecutor的执行策略。这里先给一个概述
int corePoolSize: 核心线程池大小int maximumPoolSize: 最大线程池大小long keepAliveTime, TimeUnit unit: 空闲线程最大存活时间ThreadFactory threadFactory: 线程工厂BlockingQueueRunnable workQueue: 阻塞队列RejectedExecutionHandler handler: 拒绝策略
线程池的状态和工作线程数量
在开始看任务执行前我们需要先看一个原子变量ctl这个变量隐含了两个概念一个是工作线程的数量另一个是线程池的状态。 这两个值通过或运算组合在一起其中低位保存工作线程数量高位保存状态。这里高位和低位的分隔是第29位这意味着最多只能有2^29 - 1大概500万个工作线程。
线程池的状态包含以下几种
RUNNING(运行中): 接受新任务和处理队列中的任务SHUTDOWN(): 不再接受新任务但是仍然会执行队列中的任务STOP(停止): 不接受新任务不执行队列中的任务中断正在进行的任务TIDYING(清理): 所有任务都执行完成workerCount为0线程过渡为TIDYING状态执行terminated()钩子方法TERMINATED(): terminated已经执行
状态的改变
RUNNING - SHUTDOWN: 调用shutdown()RUNNING or SHUTDOWN) - STOP: 调用shutdownNow()SHUTDOWN - TIDYING: 当队列和线程池都空了STOP - TIDYING: 当线程池空了TIDYING - TERMINATED: 当terminated方法执行完毕 private final AtomicInteger ctl new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS Integer.SIZE - 3;private static final int COUNT_MASK (1 COUNT_BITS) - 1;// runState is stored in the high-order bitsprivate static final int RUNNING -1 COUNT_BITS;private static final int SHUTDOWN 0 COUNT_BITS;private static final int STOP 1 COUNT_BITS;private static final int TIDYING 2 COUNT_BITS;private static final int TERMINATED 3 COUNT_BITS;// Packing and unpacking ctlprivate static int runStateOf(int c) { return c ~COUNT_MASK; }private static int workerCountOf(int c) { return c COUNT_MASK; }private static int ctlOf(int rs, int wc) { return rs | wc; }private static boolean runStateLessThan(int c, int s) {return c s;}private static boolean runStateAtLeast(int c, int s) {return c s;}private static boolean isRunning(int c) {return c SHUTDOWN;}private void decrementWorkerCount() {ctl.addAndGet(-1);}任务的执行
在execute概念讲解前需要先厘清两种线程池一种称为核心线程池另一种称为最大线程池。注意这里虽然是这么叫但是实际上只有一个线程池corePoolSize可以想成是一个边界。
0 ... corePoolSize ... maximumPoolSize边界里的劳工干到死除非有让他停下的理由边界外的闲了一段时间就放假。
当提交一个任务给线程池时会按以下的逻辑执行
如果工作线程数量小于corePoolSize那么核心线程池添加一个工作线程执行任务。如果线程池还在运行那么添加到工作队列。这里会有一个双重检测再判断一次线程池的状态。如果线程池不在运行状态了从队列移除执行拒绝策略。如果工作线程数量为0那么添加一个到最大线程池。如果既无法添加到线程池又无法添加到队列中执行拒绝策略。 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();}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);}else if (!addWorker(command, false))reject(command);}工作者线程
添加工作者线程有两个参数第一个Runnable firstTask意味着这个线程第一个执行的任务这个值可能是null意味着启动一个工作者线程但是不需要直接执行任务而是到队列之类的地方拿后面getTask会分析。 第二个参数意味着这个工作线程是否是添加到核心线程池。前面也说了其实只有一个池子corePoolSize只是相当于中间的边界从下面方法的实现我们也可以看出来当core为true时如果线程数量超过corePoolSize则添加失败。 private boolean addWorker(Runnable firstTask, boolean core) {retry:for (int c ctl.get();;) {// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN) (runStateAtLeast(c, STOP)|| firstTask ! null|| workQueue.isEmpty()))return false;for (;;) {if (workerCountOf(c) ((core ? corePoolSize : maximumPoolSize) COUNT_MASK))return false;if (compareAndIncrementWorkerCount(c))break retry;c ctl.get(); // Re-read ctlif (runStateAtLeast(c, SHUTDOWN))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 c ctl.get();if (isRunning(c) ||(runStateLessThan(c, STOP) firstTask null)) {if (t.getState() ! Thread.State.NEW)throw new IllegalThreadStateException();workers.add(w);workerAdded true;int s workers.size();if (s largestPoolSize)largestPoolSize s;}} finally {mainLock.unlock();}if (workerAdded) {t.start();workerStarted true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}方法的实现中前面标签retry部分是判断能不能添加后面部分是添加的实际过程。
添加一个工作者线程这个过程会从线程池中创建一个新的线程。将这个工作线程添加进HashSetWorker workers中注意本身HashSet是线程不安全的所以实际上所有对它的操作都在持有锁的情况下进行。更新largestPoolSize。启动线程。如果失败调用addWorkerFailed。
启动线程实际执行的run方法 private final class Worker extends AbstractQueuedSynchronizer implements Runnable{Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask firstTask;this.thread getThreadFactory().newThread(this);}/** Delegates main run loop to outer runWorker. */public void run() {runWorker(this);}}工作者线程不断从队列中(getTask())拿任务来执行。如果发现没有任务了就尝试自我毁灭(processWorkerExit())了。 工作者线程在执行的时候会先给自己上个排他锁所以当worker是isLocked的时候意味着正在执行任务。 任务的执行分为beforeExecute(wt, task); - task.run(); - afterExecute(task, ex);判断中断线程在beforeExecute之前。 final void runWorker(Worker w) {Thread wt Thread.currentThread();Runnable task w.firstTask;w.firstTask null;w.unlock(); // allow interruptsboolean 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 interruptif ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() runStateAtLeast(ctl.get(), STOP))) !wt.isInterrupted())wt.interrupt();try {beforeExecute(wt, task);try {task.run();afterExecute(task, null);} catch (Throwable ex) {afterExecute(task, ex);throw ex;}} finally {task null;w.completedTasks;w.unlock();}}completedAbruptly false;} finally {processWorkerExit(w, completedAbruptly);}}getTask会不断的尝试在队列中拿任务除非超时或者终止线程池或者队列空了。 private Runnable getTask() {boolean timedOut false; // Did the last poll() time out?for (;;) {int c ctl.get();// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN) (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc workerCountOf(c);// Are workers subject to culling?boolean timed allowCoreThreadTimeOut || wc corePoolSize;if ((wc maximumPoolSize || (timed timedOut)) (wc 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {Runnable r timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r ! null)return r;timedOut true;} catch (InterruptedException retry) {timedOut false;}}}拒绝策略
当线程池和阻塞队列无法执行任务时由拒绝策略决定任务何去何从。
public interface RejectedExecutionHandler {void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}接口RejectedExecutionHandler定义了rejectedExecution方法该方法接受两个参数一个是待执行的任务另一个是当前线程池。调用的地方在ThreadPoolExecutor中的 final void reject(Runnable command) {handler.rejectedExecution(command, this);}其中内置了以下几种策略
CallerRunsPolicy: 调用者线程执行AbortPolicy: 抛弃且抛出异常DiscardPolicy: 抛弃无异常DiscardOldestPolicy: 抛弃最老的任务执行当前的任务
CallerRunsPolicy 调用者线程执行
如果线程池没有shutdown则执行。 public static class CallerRunsPolicy implements RejectedExecutionHandler {public CallerRunsPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {r.run();}}}AbortPolicy 抛弃并抛异常策略
这是默认的执行方式拒绝执行并抛出一个异常。 public static class AbortPolicy implements RejectedExecutionHandler {public AbortPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {throw new RejectedExecutionException(Task r.toString() rejected from e.toString());}}DiscardPolicy 抛弃不抛异常策略
抛弃但是不做任何事情。 public static class DiscardPolicy implements RejectedExecutionHandler {public DiscardPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}}DiscardOldestPolicy 抛弃队列头的任务
抛弃队列中最老的任务然后执行当前任务。 public static class DiscardOldestPolicy implements RejectedExecutionHandler {public DiscardOldestPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {e.getQueue().poll();e.execute(r);}}}
这里在Javadoc有提到以下的实现这个实现优于DiscardOldestPolicy它会触发被抛弃任务的回调或者记录一些额外的信息 new RejectedExecutionHandler() {public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {Runnable dropped e.getQueue().poll();if (dropped instanceof Future?) {((Future?)dropped).cancel(false);// also consider logging the failure}e.execute(r); // retry线程池监控
getPoolSize(): 返回当前线程池中的线程数量getActiveCount(): 返回线程池中(大约)活跃线程的数量getLargestPoolSize(): 返回线程池中最多的时候运行的线程数量getTaskCount(): 返回(大约)提交过的任务数量completedTaskCount(): 返回(大约)执行完成的任务数量
线程池预热
prestartCoreThread: 预热一个核心线程prestartAllCoreThreads: 预热所有核心线程
设置核心线程池数量
动态设置核心线程数量如果当前核心线程数大于设置值那么中断那些空闲的线程。当前核心线程数小于设置值那么预热差值或队列大小的小值。 public void setCorePoolSize(int corePoolSize) {if (corePoolSize 0 || maximumPoolSize corePoolSize)throw new IllegalArgumentException();int delta corePoolSize - this.corePoolSize;this.corePoolSize corePoolSize;if (workerCountOf(ctl.get()) corePoolSize)interruptIdleWorkers();else if (delta 0) {// We dont really know how many new threads are needed.// As a heuristic, prestart enough new workers (up to new// core size) to handle the current number of tasks in// queue, but stop if queue becomes empty while doing so.int k Math.min(delta, workQueue.size());while (k-- 0 addWorker(null, true)) {if (workQueue.isEmpty())break;}}}动态设置最大线程池数量 public void setMaximumPoolSize(int maximumPoolSize) {if (maximumPoolSize 0 || maximumPoolSize corePoolSize)throw new IllegalArgumentException();this.maximumPoolSize maximumPoolSize;if (workerCountOf(ctl.get()) maximumPoolSize)interruptIdleWorkers();}
