北京网站优化策略,2021年最火装修风格,有限责任公司成立条件,广州网站公司建设网站1. 工作原理1.1 流程图1.2 执行示意图从上图得知如果当前运行的线程数小于corePoolSize(核心线程数)#xff0c;则会创建新线程作为核心线程来执行任务(注意#xff0c;执行这一步需要获取全局锁)。如果运行的线程等于或多于corePoolSize#xff0c;则将任务加入BlockingQue…1. 工作原理1.1 流程图1.2 执行示意图从上图得知如果当前运行的线程数小于corePoolSize(核心线程数)则会创建新线程作为核心线程来执行任务(注意执行这一步需要获取全局锁)。如果运行的线程等于或多于corePoolSize则将任务加入BlockingQueue中。如果无法将任务加入BlockingQueue(队列已满)则将创建线程来执行任务(注意执行这一步骤需要获取全局锁)。如果第三步创建线程使得当前线程数大于maximumPoolSize任务将被拒绝并调用RejectdExecutionHandler.rejectedExecution()方法。2. 源码解析2.1 核心参数corePoolSize 核心线程数池中所保存的线程数包括空闲线程。maximumPoolSize 线程池最大容量在核心线程数的基础上可能会额外增加一些非核心线程需要注意的是只有当workQueue队列填满时才会创建多于corePoolSize的线程(线程池总线程数不超过maxPoolSize)workQueue 用于保存任务的队列 SynchronousQueue 线性安全、capacity是0不存储任何元素、可以在两个线程中传递同一个对象。synchronousQueue.put(object);synchronousQueue.take();如果不希望任务在队列中等待而是希望将任务直接移交给工作线程可使用SynchronousQueue作为等待队列。LinkedBlockingQueue 链式队列队列容量不足或为0时自动阻塞。锁是分离的高并发的情况下生产者和消费者可以并行地操作队列中的数据可以提高吞吐量。当QPS很高发送数据很大大量的任务被添加到这个无界LinkedBlockingQueue 中导致cpu和内存飙升服务器挂掉。ArrayBlockingQueue 数组实现的有界队列使用独占锁生产者和消费者无法并行操作在高并发场景下会成为性能瓶颈。PriorityBlockingQueue PriorityBlockingQueue中的优先级由任务的Comparator决定。keepAliveTime 空闲线程的存活时间 当线程空闲时间达到keepAliveTime时线程会退出直到线程数量corePoolSize。如果allowCoreThreadTimeouttrue则会直到线程数量0。threadFactory 线程工厂 默认DefaultThreadFactory 创建同样分组(ThreadGroup) 同样优先级(NORM_PRIORITY) non-daemon的线程。handler 拒绝策略处理 默认AbortPolicy 抛异常RejectedExecutionExceptionAbortPolicy不执行该任务并抛出RejectedExecutionException异常。CallerRunsPolicy由调用线程处理该任务如果执行程序已关闭则会丢弃该任务。DiscardOldestPolicy丢弃队列中最老的一个任务然后重新尝试执行任务(重复此过程。DiscardPolicy不执行该任务也不抛异常。2.2 线程池状态2.2.1 线程池状态 成员变量ctl是Integer的原子变量使用一个变量同时记录线程池状态和线程池中线程个数假设计算机硬件的Integer类型是32位二进制标示如下面代码所示其中高3位用来表示线程池状态后面29位用来记录线程池线程个数。//线程个数掩码位数并不是所有平台int类型是32位所以准确说是具体平台下Integer的二进制位数-3后的剩余位数才是线程的个数private static final int COUNT_BITS Integer.SIZE - 3;
//线程最大个数(低29位)00011111111111111111111111111111
private static final int CAPACITY (1 COUNT_BITS) - 1;
//高3位11100000000000000000000000000000
private static final int RUNNING -1 COUNT_BITS;
//高3位00000000000000000000000000000000
private static final int SHUTDOWN 0 COUNT_BITS;
//高3位00100000000000000000000000000000
private static final int STOP 1 COUNT_BITS;
//高3位01000000000000000000000000000000
private static final int TIDYING 2 COUNT_BITS;
//高3位01100000000000000000000000000000
private static final int TERMINATED 3 COUNT_BITS;
获取线程池的运行状态和线程池中的线程个数
// 获取高三位 运行状态
private static int runStateOf(int c) { return c ~CAPACITY; }
//获取低29位 线程个数
private static int workerCountOf(int c) { return c CAPACITY; }
//计算ctl新值线程状态 与 线程个数
private static int ctlOf(int rs, int wc) { return rs | wc; }2.2.2 线程池状态含义RUNNING接收新任务并且处理阻塞队列里的任务。SHUTDOWN拒绝新任务但是处理阻塞队列里的任务。STOP拒绝新任务并且抛弃阻塞队列里的任务同时中断正在处理的任务。TIDYING所有任务都执行完包含阻塞队列里面任务当前线程池活动线程为0将要调用terminated方法。TERMINATED终止状态。terminated方法调用完成以后的状态。2.3 执行方法2.3.1 线程池中任务提交方法execute()方法public void execute(Runnable command) {//(1) 如果任务为null则抛出NPE异常if (command null)throw new NullPointerException();//2获取当前线程池的状态线程个数变量的组合值int c ctl.get();//3当前线程池线程个数是否小于corePoolSize,小于则开启新线程运行if (workerCountOf(c) corePoolSize) {if (addWorker(command, true))return;c ctl.get();}//4如果线程池处于RUNNING状态则添加任务到阻塞队列if (isRunning(c) workQueue.offer(command)) {//4.1二次检查int recheck ctl.get();//4.2如果当前线程池状态不是RUNNING则从队列删除任务并执行拒绝策略if (! isRunning(recheck) remove(command))reject(command);//4.3如果当前线程池线程为空则添加一个线程else if (workerCountOf(recheck) 0)addWorker(null, false);}//5如果队列满了则新增线程新增失败则执行拒绝策略else if (!addWorker(command, false))reject(command);
} submit(Runnable task)方法public Future? submit(Runnable task) {// 6 NPE判断if (task null) throw new NullPointerException();// 7 包装任务为FutureTaskRunnableFutureVoid ftask newTaskFor(task, null);// 8 投递到线程池执行execute(ftask);// 9 返回ftaskreturn ftask;
}protected T RunnableFutureT newTaskFor(Runnable runnable, T value) {return new FutureTaskT(runnable, value);
}public FutureTask(Runnable runnable, V result) {//将runnable适配为Callable类型任务并且让result作为执行结果this.callable Executors.callable(runnable, result);this.state NEW; // ensure visibility of callable
} submit(Runnable taskT result)方法public T FutureT submit(CallableT task) {if (task null) throw new NullPointerException();RunnableFutureT ftask newTaskFor(task);execute(ftask);return ftask;
}protected T RunnableFutureT newTaskFor(CallableT callable) {return new FutureTaskT(callable);
}2.3.2 线程池中任务执行方法 当用户线程提交任务到线程池后在线程池没有执行拒绝策略的情况下用户线 程会马上返回而提交的任务要么直接切换到线程池中的Worker线程来执行要么先放入线程池的阻塞队列里面稍后再由Worker线程来执行。Worker(Runnable firstTask) {// 在调用runWorker前禁止中断 避免当前Worker在调用runWorker方法前被中断当其他线程调用了线程池的shutdownNow时如果Worker状态≥0则会中断该线程。status0标示锁未被获取的状态state1标示锁已经被获取的状态state–1是创建Worker时默认的状态。setState(-1);this.firstTask firstTask;//使用线程池中指定的线程池工厂创建一个线程作为该Worker对象的执行线程this.thread getThreadFactory().newThread(this);
}public void run() {runWorker(this);//委托给runWorker方法
}final void runWorker(Worker w) {Thread wt Thread.currentThread();Runnable task w.firstTask;w.firstTask null;//(1)status设置为0允许中断。这时候调用shutdownNow会中断Worker线程。w.unlock();boolean completedAbruptly true;try {//(2)当前tasknull或者调用getTask从任务队列获取的任务返回nullwhile (task ! null || (task getTask()) ! null) {//(2.1)获取工作线程内部持有的独占锁w.lock();...try {//(2.2)任务执行前干一些事情beforeExecute(wt, task);Throwable thrown null;try {//(2.3)执行任务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 {//(2.4)任务执行完毕后干一些事情afterExecute(task, thrown);}} finally {task null;//(2.5)统计当前Worker完成了多少个任务w.completedTasks;w.unlock();}}completedAbruptly false;} finally {//(3)执行清工作processWorkerExit(w, completedAbruptly);}
}private void processWorkerExit(Worker w, boolean completedAbruptly) {...//(3.1)统计整个线程池完成的任务个数,并从工作集里面删除当前woker 加全局锁final ReentrantLock mainLock this.mainLock;mainLock.lock();try {completedTaskCount w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}//(3.2)如果当前线程池状态是shutdown状态并且工作队列为空或者当前是stop状态并且当前线程池里面没有活动线程则设置线程池状态为TERMINATED。tryTerminate();//(3.3)判断当前线程中的线程个数是否小于核心线程个数如果是则新增一个线程。int c ctl.get();if (runStateLessThan(c, STOP)) {if (!completedAbruptly) {int min allowCoreThreadTimeOut ? 0 : corePoolSize;if (min 0 ! workQueue.isEmpty())min 1;if (workerCountOf(c) min)return; // replacement not needed}addWorker(null, false);}
}2.3.3 线程池关闭方法 shutdown()方法 调用 shutdown后,线程池就不会再接收新的任务,但是工作队列里面的任务还是要执行的,该方法是立刻返回的,并不同步等待队列任务完成再返回。 首先使用CAS设置当前线程池状态为TIDYING如果成功则执行扩展接口terminated在线程池状态变为TERMINATED前做一些事情然后设置当前 线程池状态为TERMINATED最后调用termination.signalAll()来激活调用线程池的 awaitTermination系列方法被阻塞的所有线程。public void shutdown() {final ReentrantLock mainLock this.mainLock;mainLock.lock();try {//(1)权限检查如果设置了安全管理器则看当前调用shutdown命令的线程是否有关闭线程的权限如果有权限则还要看调用线程是否有中断工作线程的权限如果没有权限则抛出SecurityException或者NullPointerException异常。checkShutdownAccess();//(2)设置当前线程池状态为SHUTDOWN如果已经是SHUTDOWN则直接返回advanceRunState(SHUTDOWN);//(3)设置中断标志。这里首先加了全局锁同时只有一个线程可以调用shutdown设置中断标志。然后尝试获取Worker本身的锁获取成功则设置中断标识由于正在执行的任务已经获取了锁所以正在执行的任务没有被中断。这里中断的是阻塞到getTask()方法企图从队列里获取任务的线程也就是空闲线程。interruptIdleWorkers();onShutdown();} finally {mainLock.unlock();}//(4)尝试状态变为TERMINATEDtryTerminate();
}private void advanceRunState(int targetState) {for (;;) {int c ctl.get();if (runStateAtLeast(c, targetState) ||ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))break;}
}private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t w.thread;//如果工作线程没有被中断并且没有正在运行则设置中断if (!t.isInterrupted() w.tryLock()) {try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}
}final void tryTerminate() {for (;;) {
...int c ctl.get();
...final ReentrantLock mainLock this.mainLock;mainLock.lock();try {//设置当前线程池状态为TIDYINGif (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {try {terminated();} finally {//设置当前线程池状态为TERMINATEDctl.set(ctlOf(TERMINATED, 0));//激活调用条件变量termination的await系列方法被阻塞的所有线程termination.signalAll();}return;}} finally {mainLock.unlock();}}
} shutdownNow()方法 调用 shutdownnow后,线程池就不会再接收新的任务,并且会丢弃工作队列里面的任务,正在执行的任务也会被中断,该方法是立刻返回的,并不同步等待激活的任务执行完成再返回。 调用线程池队列的drainTo方法把队列中的任务移除到taskList 里如果发现线程池队列还不为空比如DelayQueue或者其他类型的队列drainTo可能 移除元素失败则循环移除里面的元素最后返回移除的任务列表。public ListRunnable shutdownNow() {ListRunnable tasks;final ReentrantLock mainLock this.mainLock;mainLock.lock();try {//5)权限检查checkShutdownAccess();//(6) 设置线程池状态为stopadvanceRunState(STOP);//(7)中断所有线程这里需要注意的是中断所有线程包含空闲线程和正在执行任务的线程interruptWorkers();//8移动队列任务到taskstasks drainQueue();} finally {mainLock.unlock();}//(9)终止状态tryTerminate();return tasks;
}private void interruptWorkers() {final ReentrantLock mainLock this.mainLock;mainLock.lock();try {for (Worker w : workers)w.interruptIfStarted();} finally {mainLock.unlock();}
}private ListRunnable drainQueue() {//8.1获取任务队列BlockingQueueRunnable q workQueue;ArrayListRunnable taskList new ArrayListRunnable();//8.2 从任务队列移除任务到taskList列表q.drainTo(taskList);//8.3 如果q还不为空则说明drainTo接口调用失效则循环移除if (!q.isEmpty()) {for (Runnable r : q.toArray(new Runnable[0])) {if (q.remove(r))taskList.add(r);}}//8.4返回异常的任务列表return taskList;
}2.4 参数优化2.4.1 默认值corePoolSize 1
maxPoolSize Integer.MAX_VALUE
queueCapacity Integer.MAX_VALUE
keepAliveTime 60s
allowCoreThreadTimeout false
rejectedExecutionHandler AbortPolicy()需要根据几个值来决定tasks 每秒的任务数假设为500~1000taskcost每个任务花费时间假设为0.1sresponsetime系统允许容忍的最大响应时间假设为1scorePoolSize corePoolSize 每秒需要多少个线程处理threadcount tasks/(1/taskcost) taskstaskcout (500 ~ 1000)0.1 50~100 个线程。corePoolSize设置应该大于50。根据8020原则如果80%的每秒任务数小于800那么corePoolSize设置为80。queueCapacity队列容量 queueCapacity (coreSizePool/taskcost)*responsetime切记不能设置为Integer.MAX_VALUE这样队列会很大线程数只会保持在corePoolSize大小当任务陡增时,不能新开线程来执行响应时间会随之陡增。maxPoolSize最大线程数在生产环境上我们往往设置成corePoolSize一样这样可以减少在处理过程中创建线程的开销。rejectedExecutionHandler根据具体情况来决定任务不重要可丢弃任务重要则要利用一些缓冲机制来处理。keepAliveTime和allowCoreThreadTimeout采用默认通常能满足。2.4.2 参考值Configuration
public class ConcurrentThreadGlobalConfig {Beanpublic ThreadPoolTaskExecutor defaultThreadPool() {ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();//核心线程数目executor.setCorePoolSize(65);//指定最大线程数executor.setMaxPoolSize(65);//队列中最大的数目executor.setQueueCapacity(650);//线程名称前缀executor.setThreadNamePrefix(DefaultThreadPool_);//rejection-policy当pool已经达到max size的时候如何处理新任务//CALLER_RUNS不在新线程中执行任务而是由调用者所在的线程来执行//对拒绝task的处理策略executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());//线程空闲后的最大存活时间executor.setKeepAliveSeconds(60);//加载executor.initialize();return executor;}
}以上
