11 java 线程池 实现原理

1 volatile int runState;
2 static final int RUNNING    = 0;
3 static final int SHUTDOWN   = 1;
4 static final int STOP       = 2;
5 static final int TERMINATED = 3;

private final BlockingQueue<Runnable> workQueue;              //任务缓存队列，用来存放等待执行的任务
private final ReentrantLock mainLock = new ReentrantLock();   //线程池的主要状态锁，对线程池状态（比如线程池大小runState等）的改变都要使用这个锁
private final HashSet<Worker> workers = new HashSet<Worker>();  //用来存放工作集
 
private volatile long  keepAliveTime;    //线程存活时间   
private volatile boolean allowCoreThreadTimeOut;   //是否允许为核心线程设置存活时间
private volatile int   corePoolSize;     //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int   maximumPoolSize;   //线程池最大能容忍的线程数
private volatile int   poolSize;       //线程池中当前的线程数


private volatile RejectedExecutionHandler handler; //任务拒绝策略
private volatile ThreadFactory threadFactory;   //线程工厂，用来创建线程
private int largestPoolSize;   //用来记录线程池中曾经出现过的最大线程数
private long completedTaskCount;   //用来记录已经执行完毕的任务个数

 1 public void execute(Runnable command) {
 2     if (command == null)
 3         throw new NullPointerException();
 4     if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
 5         if (runState == RUNNING && workQueue.offer(command)) {
 6             if (runState != RUNNING || poolSize == 0)
 7                 ensureQueuedTaskHandled(command);
 8         }
 9         else if (!addIfUnderMaximumPoolSize(command))
10             reject(command); // is shutdown or saturated
11     }
12 }

1 if (poolSize >= corePoolSize |!addIfUnderCorePoolSize(command))

addIfUnderCorePoolSize(command)

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
    Thread t = null;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (poolSize < corePoolSize && runState == RUNNING)
            t = addThread(firstTask);        //创建线程去执行firstTask任务   
        } finally {
        mainLock.unlock();
    }
    if (t == null)
        return false;
    t.start();
    return true;
}

private Thread addThread(Runnable firstTask) {
    Worker w = new Worker(firstTask);
    Thread t = threadFactory.newThread(w);  //创建一个线程，执行任务   
    if (t != null) {
        w.thread = t;            //将创建的线程的引用赋值为w的成员变量       
        workers.add(w);
        int nt = ++poolSize;     //当前线程数加1       
        if (nt > largestPoolSize)
            largestPoolSize = nt;
    }
    return t;
}

 1 private final class Worker implements Runnable {
 2     private final ReentrantLock runLock = new ReentrantLock();
 3     private Runnable firstTask;
 4     volatile long completedTasks;
 5     Thread thread;
 6     Worker(Runnable firstTask) {
 7         this.firstTask = firstTask;
 8     }
 9     boolean isActive() {
10         return runLock.isLocked();
11     }
12     void interruptIfIdle() {
13         final ReentrantLock runLock = this.runLock;
14         if (runLock.tryLock()) {
15             try {
16         if (thread != Thread.currentThread())
17         thread.interrupt();
18             } finally {
19                 runLock.unlock();
20             }
21         }
22     }
23     void interruptNow() {
24         thread.interrupt();
25     }
26  
27     private void runTask(Runnable task) {
28         final ReentrantLock runLock = this.runLock;
29         runLock.lock();
30         try {
31             if (runState < STOP &&
32                 Thread.interrupted() &&
33                 runState >= STOP)
34             boolean ran = false;
35             beforeExecute(thread, task);   //beforeExecute方法是ThreadPoolExecutor类的一个方法，没有具体实现，用户可以根据
36             //自己需要重载这个方法和后面的afterExecute方法来进行一些统计信息，比如某个任务的执行时间等           
37             try {
38                 task.run();
39                 ran = true;
40                 afterExecute(task, null);
41                 ++completedTasks;
42             } catch (RuntimeException ex) {
43                 if (!ran)
44                     afterExecute(task, ex);
45                 throw ex;
46             }
47         } finally {
48             runLock.unlock();
49         }
50     }
51  
52     public void run() {
53         try {
54             Runnable task = firstTask;
55             firstTask = null;
56             while (task != null || (task = getTask()) != null) {
57                 runTask(task);
58                 task = null;
59             }
60         } finally {
61             workerDone(this);   //当任务队列中没有任务时，进行清理工作       
62         }
63     }
64 }

public void run() {
    try {
        Runnable task = firstTask;
        firstTask = null;
        while (task != null || (task = getTask()) != null) {
            runTask(task);
            task = null;
        }
    } finally {
        workerDone(this);
    }
}

 1 Runnable getTask() {
 2     for (;;) {
 3         try {
 4             int state = runState;
 5             if (state > SHUTDOWN)
 6                 return null;
 7             Runnable r;
 8             if (state == SHUTDOWN)  // Help drain queue
 9                 r = workQueue.poll();
10             else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间，
11                 //则通过poll取任务，若等待一定的时间取不到任务，则返回null
12                 r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
13             else
14                 r = workQueue.take();
15             if (r != null)
16                 return r;
17             if (workerCanExit()) {    //如果没取到任务，即r为null，则判断当前的worker是否可以退出
18                 if (runState >= SHUTDOWN) // Wake up others
19                     interruptIdleWorkers();   //中断处于空闲状态的worker
20                 return null;
21             }
22             // Else retry
23         } catch (InterruptedException ie) {
24             // On interruption, re-check runState
25         }
26     }
27 }

 1 private boolean workerCanExit() {
 2     final ReentrantLock mainLock = this.mainLock;
 3     mainLock.lock();
 4     boolean canExit;
 5     //如果runState大于等于STOP，或者任务缓存队列为空了
 6     //或者  允许为核心池线程设置空闲存活时间并且线程池中的线程数目大于1
 7     try {
 8         canExit = runState >= STOP ||
 9             workQueue.isEmpty() ||
10             (allowCoreThreadTimeOut &&
11              poolSize > Math.max(1, corePoolSize));
12     } finally {
13         mainLock.unlock();
14     }
15     return canExit;
16 }

 1 void interruptIdleWorkers() {
 2     final ReentrantLock mainLock = this.mainLock;
 3     mainLock.lock();
 4     try {
 5         for (Worker w : workers)  //实际上调用的是worker的interruptIfIdle()方法
 6             w.interruptIfIdle();
 7     } finally {
 8         mainLock.unlock();
 9     }
10 }

 1 void interruptIfIdle() {
 2     final ReentrantLock runLock = this.runLock;
 3     if (runLock.tryLock()) {    //注意这里，是调用tryLock()来获取锁的，因为如果当前worker正在执行任务，锁已经被获取了，是无法获取到锁的
 4                                 //如果成功获取了锁，说明当前worker处于空闲状态
 5         try {
 6     if (thread != Thread.currentThread())  
 7     thread.interrupt();
 8         } finally {
 9             runLock.unlock();
10         }
11     }
12 }

private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
    Thread t = null;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (poolSize < maximumPoolSize && runState == RUNNING)
            t = addThread(firstTask);
    } finally {
        mainLock.unlock();
    }
    if (t == null)
        return false;
    t.start();
    return true;
}

public boolean prestartCoreThread() {
    return addIfUnderCorePoolSize(null); //注意传进去的参数是null
}
 
public int prestartAllCoreThreads() {
    int n = 0;
    while (addIfUnderCorePoolSize(null))//注意传进去的参数是null
        ++n;
    return n;
}

1 ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
2 ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
3 ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
4 ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务