并发编程之线程池原理

// Executor.java
// 从线程池中分配一个线程执行任务
void execute(Runnable command);		

// ExecutorService.java
// 关闭线程池,会将已提交的任务执行完毕再关闭线程池,不接受新的提交任务
void shutdown();

// 立即关闭,不论是否有任务正在执行还是已提交未执行,都立即退出jvm不再执行,

// 主要用于处理定时任务
//ScheduledExecutorService.java
// 创建并执行一个一次性定时任务,指定在 delay&unit(时长+时间单位) 时间点将会执行
public ScheduledFuture<?> schedule(Runnable command,
                                   long delay, TimeUnit unit);
public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                   long delay, TimeUnit unit)


// 下面两个方法主要是创建并执行一个周期性任务,如果发生异常会退出程序
// 区别在于
// scheduleAtFixedRate执行周期任务之后,如果任务超过指定的周期时间,下一个任务会立刻执行不会等待
// scheduleWithFixedDelay执行任务超过周期时间,仍然会等待delay时间再进行下一个任务的执行
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                 long initialDelay,
                                                 long delay,
                                                 TimeUnit unit);

非定时线程池实现类: ThreadPoolExecutor & ForkJoinPool & FinalizableDelegatedExecutorService
定时线程池实现类:  ScheduledThreadPoolExecutor & DelegatedScheduledExecutorService

// Executors.java
// 创建一个基于无界的链表阻塞队列(LinkedBlockingQueue)且为固定线程数为nThreads的线程池
// 无界: 队列没有指定容器大小,可以不断添加元素
public static ExecutorService newFixedThreadPool(int nThreads) {}
public static ExecutorService newFixedThreadPool(int nThreads,
                                                 ThreadFactory threadFactory){}


// 创建一个基于无界的同步队列(SynchronousQueue)且线程数无限制(0 - Integer.MAX_VALUE)的线程池
// 1.该线程池核心线程数为0且线程空闲时间为60s,意味着任务执行完成或者是超时将会销毁线程
// 2.不存储数据的同步队列(最多只有一个元素的队列),仅作为缓存作用
// 即等待内部创建工作线程完成之后就立即交由线程进行消费
// 3.适用于处理任务但是不确定线程个数的场景,同时为了防止不断创建线程造成CPU资源消耗过多
// 一般会自定义添加对应的线程最大数而不是Integer.MAX_VALUE
public static ExecutorService newCachedThreadPool() {}
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {}


// 创建一个基于延迟的无界任务队列且能够执行定时任务的线程池
// 线程池的核心数量corePoolSize,最大线程数为Integer.MAX_VALUE
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize){}
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize,
                                                    ThreadFactory threadFactory)


// 创建一个基于延迟的无界任务队列,能够执行定时任务且只有一个线程的线程池
// 当该线程池中的线程被中断或者异常退出的时候,线程池会新创建一个线程继续执行后续的任务
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {}
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory){}

public ThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            ThreadFactory threadFactory,
                            RejectedExecutionHandler handler){
         
}

public void execute(Runnable command) {
     if (command == null)
         throw new NullPointerException();
// ctl 高三位表示线程池状态,低(Integer.SIZE - 3)位表示线程个数,也就是ctl = 线程状态 + 线程个数
// 获取当前池的状态和线程个数的组合值
     int c = ctl.get();
	
// 当前线程的个数是否小于核心线程数
     if (workerCountOf(c) < corePoolSize) {
     	 // 创建新的线程执行任务,会通过加锁的方式保存工作线程,如果成功则执行任务
         if (addWorker(command, true))
             return;
         c = ctl.get();
     }
     // 判断当前线程池的状态是否为Running,是则将任务加入阻塞队列中,说明上述的corePool不够用
     if (isRunning(c) && workQueue.offer(command)) {
         int recheck = ctl.get();
         if (! isRunning(recheck) && remove(command))
         	 // 线程池已经shutdown并且不接受当前的任务,交由指定的策略处理
             reject(command);
         else if (workerCountOf(recheck) == 0)
         	// 线程池已经没有线程,由于任务已经加入阻塞队列,这里只需要新创建一个线程
             addWorker(null, false);
     }
     // 队列满,从线程池中创建新线程后处理任务,如果创建失败则执行策略
     else if (!addWorker(command, false))
         reject(command);
 }

// 存储线程池状态以及工作线程个数(RUNNING, 0)
// rs表示状态,wc表示工作线程个数
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
// 状态存储在对应数值的高三位
// 接收新任务并能处理队列里的任务
private static final int RUNNING    = -1 << COUNT_BITS;
// 不再接收新任务但能处理队列里的任务
private static final int SHUTDOWN   =  0 << COUNT_BITS;	
// 将会在处理任务过程被中断,不再接收和处理队列的任务
private static final int STOP       =  1 << COUNT_BITS;	
// 所有任务已经执行完毕,工作线程个数为0且在调用terminated()之前的状态
private static final int TIDYING    =  2 << COUNT_BITS;
// terminated()执行之后的状态
private static final int TERMINATED =  3 << COUNT_BITS;

// Packing and unpacking ctl
// 获取当前线程池状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 获取当前线程池的工作线程个数
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 核心代码
addWorker(Runnable firstTask, boolean core){
// ..
// 自旋检测线程个数与队列容量,corePool,maxPool的比较
	mainLock.lock();
int s = workers.size();
    if (s > largestPoolSize)
        largestPoolSize = s;
   // ...
   mainLock.unlock();
}
public int getLargestPoolSize() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        return largestPoolSize;
    } finally {
        mainLock.unlock();
    }
}

public int getPoolSize() {
     final ReentrantLock mainLock = this.mainLock;
     mainLock.lock();
     try {
         // Remove rare and surprising possibility of
         // isTerminated() && getPoolSize() > 0
         return runStateAtLeast(ctl.get(), TIDYING) ? 0
             : workers.size();
     } finally {
         mainLock.unlock();
     }
 }

public int getActiveCount() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        int n = 0;
        for (Worker w : workers)
        	// worker是基于AQS实现的,内部启动执行任务的时候会加锁
            if (w.isLocked())
                ++n;
        return n;
    } finally {
        mainLock.unlock();
    }
}

public long getTaskCount() {
     final ReentrantLock mainLock = this.mainLock;
     mainLock.lock();
     try {
         long n = completedTaskCount;	// 线程池执行完成的任务个数
         for (Worker w : workers) {
             n += w.completedTasks;   // 当前工作线程执行完成的任务个数
             if (w.isLocked())
                 ++n;
         }
         return n + workQueue.size();
     } finally {
         mainLock.unlock();
     }
 }

public long getCompletedTaskCount() {
     final ReentrantLock mainLock = this.mainLock;
     mainLock.lock();
     try {
         long n = completedTaskCount;
         for (Worker w : workers)
             n += w.completedTasks;
         return n;
     } finally {
         mainLock.unlock();
     }
 }