简单看看ThreadPoolExecutor原理(5)

当前位置:

首页 > temp > 简明python教程 >

简单看看ThreadPoolExecutor原理(5)

(w, completedAbruptly); } } //执行清理工作 private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); //获取锁，统计整个线程池完成任务的个数，由于在工作集Workers中当前的Worker已经执行完毕，就从集合中删除当前的Worker final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; workers.remove(w); } finally { //释放锁 mainLock.unlock(); } //尝试将线程池状态设置为TERMINATED，如果当前线程池状态为SHUTDOWN并且工作队列为空，或者是STOP状态，则 //可以断定当前线程池中没有活动线程，就把当前线程池状态设置为TERMINATED //在这个方法中会调用termination.signalAll()方法唤醒一些阻塞的线程，这些阻塞的线程是由于调用了线程池 //的awaitTermination方法被阻塞的 tryTerminate(); 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); } }

　　我们再看看reject是怎么执行拒绝策略的，就以默认的AbortPolicy策略为例子，其实就是抛出异常：

final void reject(Runnable command) {
    handler.rejectedExecution(command, this);
}

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());
    }
}

五.shutdown方法

　　在最上面使用线程池的例子中，我们在最后调用了线程池的shutdown方法了，这表示当前线程池不再接收新的任务了，但是当前线程池中任务队列中还是要执行的；

public void shutdown() {
    //获取锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //权限检查
        checkShutdownAccess();
        //设置当前线程池状态为SHUTDOWN，如果当前线程池状态已经是SHUTDOWN那就直接返回
        advanceRunState(SHUTDOWN);
        //设置所有空闲线程的中断标志
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    //尝试将线程池状态设置为TERMINATED
    tryTerminate();
}

//权限检查，由于当前线程调用了shutdown()方法，于是要先检查当前线程有没有权限关闭线程池
//如果有关闭线程池的权限，还要检查是否有中断工作线程的权限
private void checkShutdownAccess() {
    SecurityManager security = System.getSecurityManager();
    if (security != null) {
        security.checkPermission(shutdownPerm);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)
                security.checkAccess(w.thread);
        } finally {
            mainLock.unlock();
        }
    }
}

//shutdown方法中设置当前线程池状态为SHUTDOWN
//可以看到当线程池状态如果>=SHUTDOWN就直接返回，如果不是SHUTDOWN状态，那就CAS设置成SHUTDOWN状态
private void advanceRunState(int targetState) {
    for (;;) {
        int c = ctl.get();
        if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

//

栏目列表