Java - ThreadPoolExecutor线程池分析 - JAVA

TOP

Java - ThreadPoolExecutor线程池分析(三)

2023-09-09 10:25:58 【大中小】浏览:114次

// 判断线程池的状态是否大于等于SHUTDOWN，满足则说明线程不是RUNNING状态 if (rs >= SHUTDOWN && // 如果这三个条件都满足，就代表是要添加非核心工作线程去处理阻塞队列中任务 // 如果三个条件有一个不满足，返回false配合！，就不需要添加 ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) // 不需要添加工作线程 return false; for (;;) { // ===============工作线程个数判断========================== // 基于ctl拿到低29位的值，代表工作线程的个数 int wc = workerCountOf(c); // 如果工作线程个数大于最大工作线程数CAPACITY值，就不可以添加，返回false if (wc >= CAPACITY || // 基于core来判断添加的是否是核心工作线程 // 如果是核心：基于corePoolSize去判断 // 如果是非核心：基于maximumPoolSize去判断 wc >= (core ? corePoolSize : maximumPoolSize)) // 代表不能添加，工作线程个数不满足要求 return false; // 针对ctl + 1 ，采用CAS操作 if (compareAndIncrementWorkerCount(c)) // CAS成功后，直接退出外层循环，代表可以进行执行添加工作线程操作了 break retry; // 重新获取一次ctl值 c = ctl.get(); // Re-read ctl // 判断重新获取的ctl中，线程池的状态与之前是否有区别 // 如果状态发生改变，需要重新去判断线程状态 if (runStateOf(c) != rs) // 重新进入外层for循环 continue retry; // else CAS failed due to workerCount change; retry inner loop } } // =======================第二块==================== // 添加工作线程以及启动工作线程 // 声明了三个变量 // workerStarted 表示工作线程启动状态，默认false boolean workerStarted = false; // workerAdded 表示工作线程添加状态，默认false boolean workerAdded = false; // w 表示工作线程 Worker w = null; try { // 构建工作线程，并且将任务传递进去 w = new Worker(firstTask); // 获取worker中的thread对象 final Thread t = w.thread; // 判断thread是否不为null，在new worker时，内部会通过给予的threadFactory去构造thread交给worker // 一般如果为null，代表ThreadFactory有问题 if (t != null) { // 加锁，保证使用worker成员变量以及对largestPoolSize赋值时，保证线程安全 final ReentrantLock mainLock = this.mainLock; // 加锁，因为后续要操作HashSet是线程不安全的 mainLock.lock(); try { // 再次获取线程池状态 int rs = runStateOf(ctl.get()); // 再次判断 // 如果满足 rs < SHUTDOWN，说明线程池是RUNNING状态，可以继续执行 // 如果线程池状态为SHUTDOWN，并且firstTask为null，添加非核心线程处理阻塞队列任务 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { // 进入这里，就可以添加工作线程 // 将threadFactory构建线程后，不能直接启动线程，如果启动则抛出异常 if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); /** * 包含池中所有工作线程的集合。只有在获得mainLock时才能访问 * private final HashSet<Worker> workers = new HashSet<Worker>(); * 将创建好的worker放入工作线程集合中 */ workers.add(w); // 获取工作线程集合的大小，拿到工作线程个数 int s = workers.size(); // largestPoolSize在记录最大线程个数的记录 // 如果当前工作线程个数，大于最大线程个数的记录，就赋值 if (s > largestPoolSize) largestPoolSize = s; // 设置工作线程添加成功 workerAdded = true; } } finally { // 释放锁 mainLock.unlock(); } // 如果工作线程添加成功 if (workerAdded) { // 直接启动worker中的线程 t.start(); // 设置启动工作线程成功 workerStarted = true; } } } finally { // 做补偿的操作，如果工作线程启动失败，将这个添加失败的工作线程处理掉 if (! workerStarted) // 从工作线程的集合中移除掉 addWorkerFailed(w); } // 返回工作线程释放启动成功 return workerStarted; }

addWorker方法中的Worker对象

线程的执行是交接Worker对象来做的，Worker对象中比较核心的就是基于AQS来实现的线程中断

private final class Worker
    extends AbstractQueuedSynchronizer      //  基于AQS来实现线程中断
    implements Runnable                     //  存储需要执行的任务
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** 工作线程的Thread对象，初始化时构建出来 */
    final Thread thread;
    /** 需要执行的任务 */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {

首页上一页 1 2 3 4 5 6 下一页尾页 3/6/6
【大中小】【打印】【繁体】【投稿】【收藏】【推荐】【举报】【评论】【关闭】【返回顶部】

上一篇：简述Spring Cache缓存策略	下一篇：14、Spring之基于注解的声明式事务