简单的美食网站模板免费下载,嘉兴市城乡与建设局网站,磐石网站seo,下载应用转载自 Java 线程池框架核心代码分析前言
多线程编程中#xff0c;为每个任务分配一个线程是不现实的#xff0c;线程创建的开销和资源消耗都是很高的。线程池应运而生#xff0c;成为我们管理线程的利器。Java 通过Executor接口#xff0c;提供了一种标准的方法将任务的提…转载自 Java 线程池框架核心代码分析前言
多线程编程中为每个任务分配一个线程是不现实的线程创建的开销和资源消耗都是很高的。线程池应运而生成为我们管理线程的利器。Java 通过Executor接口提供了一种标准的方法将任务的提交过程和执行过程解耦开来并用Runnable表示任务。
下面我们来分析一下 Java 线程池框架的实现ThreadPoolExecutor。
下面的分析基于JDK1.7
生命周期
ThreadPoolExecutor中使用CAPACITY的高3位来表示运行状态分别是
RUNNING接收新任务并且处理任务队列中的任务SHUTDOWN不接收新任务但是处理任务队列的任务STOP不接收新任务不处理任务队列同时中断所有进行中的任务TIDYING所有任务已经被终止工作线程数量为 0到达该状态会执行terminated()TERMINATEDterminated()执行完毕ThreadPoolExecutor中用原子类来表示状态位
1private final AtomicInteger ctl new AtomicInteger(ctlOf(RUNNING, 0));线程池模型
核心参数
corePoolSize最小存活的工作线程数量如果设置allowCoreThreadTimeOut那么该值为 0maximumPoolSize最大的线程数量受限于CAPACITYkeepAliveTime对应线程的存活时间时间单位由TimeUnit指定workQueue工作队列存储待执行的任务RejectExecutionHandler拒绝策略线程池满后会触发
线程池的最大容量CAPACITY中的前三位用作标志位也就是说工作线程的最大容量为(2^29)-1
四种模型
CachedThreadPool一个可缓存的线程池如果线程池的当前规模超过了处理需求时那么将回收空闲的线程当需求增加时则可以添加新的线程线程池的规模不存在任何的限制。FixedThreadPool一个固定大小的线程池提交一个任务时就创建一个线程直到达到线程池的最大数量这时线程池的大小将不再变化。SingleThreadPool一个单线程的线程池它只有一个工作线程来执行任务可以确保按照任务在队列中的顺序来串行执行如果这个线程异常结束将创建一个新的线程来执行任务。ScheduledThreadPool一个固定大小的线程池并且以延迟或者定时的方式来执行任务类似于Timer。
执行任务 execute
核心逻辑
1. 当前线程数量 corePoolSize直接开启新的核心线程执行任务addWorker(command, true)2. 当前线程数量 corePoolSize且任务加入工作队列成功
检查线程池当前状态是否处于RUNNING如果否则拒绝该任务如果是判断当前线程数量是否为 0如果为 0就增加一个工作线程。
3. 开启普通线程执行任务addWorker(command, false)开启失败就拒绝该任务
从上面的分析可以总结出线程池运行的四个阶段
poolSize corePoolSize 且队列为空此时会新建线程来处理提交的任务poolSize corePoolSize此时提交的任务进入工作队列工作线程从队列中获取任务执行此时队列不为空且未满。poolSize corePoolSize并且队列已满此时也会新建线程来处理提交的任务但是poolSize maxPoolSizepoolSize maxPoolSize并且队列已满此时会触发拒绝策略
拒绝策略
前面我们提到任务无法执行会被拒绝RejectedExecutionHandler是处理被拒绝任务的接口。下面是四种拒绝策略。
AbortPolicy默认策略终止任务抛出RejectedExceptionCallerRunsPolicy在调用者线程执行当前任务不抛异常DiscardPolicy 抛弃策略直接丢弃任务不抛异常DiscardOldersPolicy抛弃最老的任务执行当前任务不抛异常
线程池中的 Worker
Worker继承了AbstractQueuedSynchronizer和Runnable前者给Worker提供锁的功能后者执行工作线程的主要方法runWorker(Worker w)从任务队列捞任务执行。Worker 引用存在workers集合里面用mainLock守护。
12private final ReentrantLock mainLock new ReentrantLock();private final HashSetWorker workers new HashSetWorker();核心函数 runWorker
下面是简化的逻辑注意每个工作线程的run都执行下面的函数
12345678910111213final void runWorker(Worker w) { Thread wt Thread.currentThread(); Runnable task w.firstTask; w.firstTask null; while (task ! null || (task getTask()) ! null) { w.lock(); beforeExecute(wt, task); task.run(); afterExecute(task, thrown); w.unlock(); } processWorkerExit(w, completedAbruptly);}从getTask()中获取任务锁住 worker执行beforeExecute(wt, task)这是ThreadPoolExecutor提供给子类的扩展方法运行任务如果该worker有配置了首次任务则先执行首次任务且只执行一次。执行afterExecute(task, thrown);解锁 worker如果获取到的任务为 null关闭 worker
获取任务 getTask
线程池内部的任务队列是一个阻塞队列具体实现在构造时传入。
1private final BlockingQueueRunnable workQueue;getTask()从任务队列中获取任务支持阻塞和超时等待任务四种情况会导致返回null让worker关闭。
现有的线程数量超过最大线程数量线程池处于STOP状态线程池处于SHUTDOWN状态且工作队列为空线程等待任务超时且线程数量超过保留线程数量
核心逻辑根据timed在阻塞队列上超时等待或者阻塞等待任务等待任务超时会导致工作线程被关闭。
1234timed allowCoreThreadTimeOut || wc corePoolSize;Runnable r timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();在以下两种情况下等待任务会超时
允许核心线程等待超时即allowCoreThreadTimeOut(true)当前线程是普通线程此时wc corePoolSize
工作队列使用的是BlockingQueue这里就不展开了后面再写一篇详细的分析。
总结
ThreadPoolExecutor基于生产者-消费者模式提交任务的操作相当于生产者执行任务的线程相当于消费者。Executors提供了四种基于ThreadPoolExecutor构造线程池模型的方法除此之外我们还可以直接继承ThreadPoolExecutor重写beforeExecute和afterExecute方法来定制线程池任务执行过程。使用有界队列还是无界队列需要根据具体情况考虑工作队列的大小和线程的数量也是需要好好考虑的。拒绝策略推荐使用CallerRunsPolicy该策略不会抛弃任务也不会抛出异常而是将任务回退到调用者线程中执行。
