Java线程池

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
}

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

ExecutorService.execute(Runnable runable)；

FutureTask task = ExecutorService.submit(Runnable runnable);
FutureTask<T> task = ExecutorService.submit(Runnable runnable,T Result);
FutureTask<T> task = ExecutorService.submit(Callable<T> callable);

public class MyThread extends Thread {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " is running.");
    }
}
 
public class TestSingleThreadExecutor {
    public static void main(String[] args) {
        //创建一个可重用固定线程数的线程池
        ExecutorService pool = Executors.newFixedThreadPool(2);
        //创建实现了Runnable接口对象
        Thread tt1 = new MyThread();
        Thread tt2 = new MyThread();
        Thread tt3 = new MyThread();
        Thread tt4 = new MyThread();
        Thread tt5 = new MyThread();
        //将线程放入池中并执行
        pool.execute(tt1);
        pool.execute(tt2);
        pool.execute(tt3);
        pool.execute(tt4);
        pool.execute(tt5);
        //关闭
        pool.shutdown();
    }
}

private final class Worker implements Runnable {
 
    final Thread thread;
 
    Runnable firstTask;
 
    Worker(Runnable firstTask) {
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
 
    public void run() {
        runWorker(this);
    }
 
    final void runWorker(Worker w) {
        Runnable task = w.firstTask;
        w.firstTask = null;
        while (task != null || (task = getTask()) != null){
             task.run();
        }
    }
  ……
}

private Runnable getTask() {
  if(一些特殊情况) {
    return null;
  }
  Runnable r = workQueue.take();
  return r;
}

public void execute(Runnable command) {
    if (command == null) throw new NullPointerException();
    int c = ctl.get();
    // 当前线程数 < corePoolSize
    if (workerCountOf(c) < corePoolSize) {
        // 直接启动新的线程。
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 活动线程数 >= corePoolSize
    // runState为RUNNING && 队列未满
    if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();
    // 再次检验是否为RUNNING状态
    // 非RUNNING状态 则从workQueue中移除任务并拒绝
    if (!isRunning(recheck) && remove(command))
      reject(command);
    // 采用线程池指定的策略拒绝任务
    // 两种情况：
    // 1.非RUNNING状态拒绝新的任务
    // 2.队列满了启动新的线程失败（workCount > maximumPoolSize）
    } else if (!addWorker(command, false))
        reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {
    int wc = workerCountOf(c);
    if (wc >= (core ? corePoolSize : maximumPoolSize)) {
        return false;
    }
    w = new Worker(firstTask);
    final Thread t = w.thread;
    t.start();
}

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

//得到线程的状态
private static int runStateOf(int c) { return c & ~CAPACITY; }

//得到Worker的的数量
private static int workerCountOf(int c) { return c & CAPACITY; }

// 判断线程是否在运行
private static boolean isRunning(int c) { return c < SHUTDOWN; }

public abstract class AbstractExecutorService implements ExecutorService {
 
}

public class ThreadPoolExecutor extends AbstractExecutorService {
    /**
     * 整个线程池的状态控制类 ctl, 是一个 AtomicInteger ，封装了下面2个部分:
     * workerCount 表示有效的线程数
     * runState 表示线程池当前状态，是running, shutdown还是其他状态
     *
     * runState表示了线程池的整个整个生命周期，可以取以下值:
     * RUNNING: 接受新的task，处理队列中的task
     * SHUTDOWN: 不接受新的 task 但会处理队列中的 task
     * STOP: 不接受新的 task, 不处理队列中的 task, 中断正在执行的 task
     * TIDYING: 所有 task 执行结束, workerCount 是 0, 线程过渡到TIDYING会调用 terminated()钩子方法
     * TERMINATED: terminated()执行完成
     *
     * 状态转换关系:
     *
     * RUNNING -> SHUTDOWN(调用shutdown())
     * On invocation of shutdown(), perhaps implicitly in finalize()
     * (RUNNING or SHUTDOWN) -> STOP(调用shutdownNow())
     * On invocation of shutdownNow()
     * SHUTDOWN -> TIDYING(queue和pool均empty)
     * When both queue and pool are empty
     * TIDYING -> TERMINATED(调用terminated())
     * When the terminated() hook method has completed
     */
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
 
    /**
     * task 的队列
     */
    private final BlockingQueue<Runnable> workQueue;
 
    /**
     * worker 线程的 Set, 只有持有 mainlock 时才能访问
     */
    private final ReentrantLock mainLock = new ReentrantLock();
    private final HashSet<Worker> workers = new HashSet<Worker>();
 
    private final Condition termination = mainLock.newCondition();
    private int largestPoolSize;
    private long completedTaskCount;
 
    private volatile ThreadFactory threadFactory;
    private volatile RejectedExecutionHandler handler;
 
    /**
     * idle 线程waiting for work的超时时间
     */
    private volatile long keepAliveTime;
    /**
     * 如果是 false,core threads将会保持 alive 即使处于 idel 状态
     * 如果是 true,core threads会keepAliveTime作为超时时间 wait for work
     */
    private volatile boolean allowCoreThreadTimeOut;
 
    private volatile int corePoolSize;
    private volatile int maximumPoolSize;
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();
}

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;
private static final int TIDYING    =  2 << COUNT_BITS;
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; }

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

boolean workerStarted = false; //标记 worker 开启状态
boolean workerAdded = false; //标记 worker 添加状态
Worker w = null;
try {
    w = new Worker(firstTask); //将这个任务作为 worker 的第一个任务传入
    final Thread t = w.thread; //通过 worker 获取到一个线程
    if (t != null) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // Recheck while holding lock.
            // Back out on ThreadFactory failure or if
            // shut down before lock acquired.
            int rs = runStateOf(ctl.get());
            // running状态，或者 shutdown 状态但是没有新的任务
            if (rs < SHUTDOWN ||
                (rs == SHUTDOWN && firstTask == null)) {
                if (t.isAlive()) // precheck that t is startable
                    throw new IllegalThreadStateException();
                // 将这个 worker 添加到线程池中
                workers.add(w);
                int s = workers.size();
                if (s > largestPoolSize)
                    largestPoolSize = s;
                // 标记worker添加成功
                workerAdded = true;
            }
        } finally {
            mainLock.unlock();
        }
        // 如果 worker 创建成功，开启线程
        if (workerAdded) {
            t.start();
            workerStarted = true;
        }
    }
} finally {
    if (! workerStarted)
        addWorkerFailed(w);
}
return workerStarted;

private final class Worker
       extends AbstractQueuedSynchronizer
       implements Runnable {
 
    final Thread thread;
    /** Initial task to run. Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;
}

// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
public void lock()        { acquire(1); }
public boolean tryLock()  { return tryAcquire(1); }
public void unlock()      { release(1); }
public boolean isLocked() { return isHeldExclusively(); }

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

public void run() {
    runWorker(this);
}
 
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask; // 获取到 worker 的第一个任务
    w.firstTask = null;
    w.unlock(); // 标记为闲置，还没有开始任务 允许打断
    boolean completedAbruptly = true; // 异常退出标记
    try {
        // 循环取出任务，如果第一个任务不为空，或者从队列中拿到了任务
        // 只要这两个条件满足，会一直循环，直到没有任务，正常退出，或者异常退出
        while (task != null || (task = getTask()) != null) {
            w.lock();// 该线程标记为非闲置
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted. This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            // 翻译注释：1.如果线程池STOPPING状态，需要中断线程
            // 2.Thread.interrupted()是一个native方法，返回当前线程是否有被等待中断的请求
            // 3.第二个条件成立时，检查线程池状态，如果为STOP，并且没有被中断，则中断线程
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            // 执行任务
            try {
                beforeExecute(wt, task);// 执行前
                Throwable thrown = null;
                try {
                    task.run(); // 执行任务
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown); // 执行结束
                }
            } finally {
                task = null; // 将 worker 的任务置空
                w.completedTasks++;
                w.unlock(); // 释放锁，进入闲置状态
            }
        }// 循环结束
        completedAbruptly = false; // 标记为正常退出
    } finally {
      // 干掉 worker
        processWorkerExit(w, completedAbruptly);
    }
}

private Runnable getTask() {
    // 标记取任务是否超时
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // Check if queue empty only if necessary.
        // 如果线程池状态为 STOP 或者 SHUTDOWN 并且队列已经为空，回收 wroker
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        //获取当前有效线程数
        int wc = workerCountOf(c);
        // Are workers subject to culling?
        // timed 用来标记当前的 worker 是否设置超时时间，
        // 还记得获取线程池的时候 可以设置核心线程超时时间
        //1.允许核心线程超时回收(即所有线程) 2.当前有效线程超过核心线程数(需要回收)
        // 如果timed == false 则该worker不会被回收，如果没有取到任务 会一直阻塞
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
        // 回收线程条件
        // 1. 有效线程数已经大于了线程池的最大线程数或者设置了超时回收并且已经超时
        // 2. 有效线程数大于1或者队列任务已经为空
        // 只有当上面1和2 同时满足时 则试图回收线程
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
          // 如果减少workercount成功 直接回收
            if (compareAndDecrementWorkerCount(c))
                return null;
          // 否则重走循环，从第一个判断条件处回收
            continue;
        }
        // 取任务
        try {
          // 根据是否设置超时回收来选择不同的取任务的方式
          // poll 方法取任务会有超时时间，超过时间则返回null
          // take 方法没有超时时间，阻塞式方法
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
          // 如果任务不为空返回任务
            if (r != null)
                return r;
          // 否则标记超时 进入下一次循环等待回收
            timedOut = true;
        } catch (InterruptedException retry) {
          // 如果出现异常，试图重试
            timedOut = false;
        }
    }
}

public E take() throws InterruptedException {
    E x; // 任务
    int c = -1; // 取出任务后的剩余任务数量
    final AtomicInteger count = this.count; // 当前任务数量
    final ReentrantLock takeLock = this.takeLock; // 加锁防止并发
    takeLock.lockInterruptibly();
    try {
      // 如果队列数量为空，则一直循环，阻塞线程
        while (count.get() == 0) {
            notEmpty.await();
        }
      // 取出任务
        x = dequeue();
      // 任务数量减一
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();// 标记队列非空
    } finally {
        takeLock.unlock(); // 释放锁
    }
    if (c == capacity)
        signalNotFull();//标记队列已满
    return x;// 返回任务
}

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    E x = null; // 存放的任务
    int c = -1;
    long nanos = unit.toNanos(timeout); // 超时时间
    final AtomicInteger count = this.count; // 队列中的数量
    final ReentrantLock takeLock = this.takeLock; // 加锁防止并发
    takeLock.lockInterruptibly();
    try {
        // 如果队列为空，则不断的循环
        while (count.get() == 0) {
            // 如果当倒计时小于0 即超时时间到 则返回空
            if (nanos <= 0)
                return null;
            // 让线程等待
            nanos = notEmpty.awaitNanos(nanos);
        }
        x = dequeue(); // 取出一个任务
        c = count.getAndDecrement(); // 取出后的队列数量
        if (c > 1)
            notEmpty.signal(); // 标记非空
    } finally {
        takeLock.unlock(); // 释放锁
    }
    if (c == capacity)
        signalNotFull(); // 标记队列已满
    return x; // 返回任务
}

processWorkerExit(w, completedAbruptly);

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();
    // 加锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    // 记录线程池完成的任务总数，从 workers 中移除该 worker
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
 
    tryTerminate(); // 尝试关闭池子
    int c = ctl.get();
    // 以下的代码是判断需不需要给线程池创建一个新的线程
    // 如果线程池的状态是 RUNNING 或者 SHUTDOWN 进一步判断需不需要创建
    if (runStateLessThan(c, STOP)) {
        // 如果为异常退出直接创建，如果不是异常退出进入判断
        if (!completedAbruptly) {
            // 获取线程池应该存在的最小线程数 如果设置了超时 则是0，否则是核心线程数
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            // 如果 min 是0 但是队列又不为空，则 min 应该是1
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            //如果当前池中的有效线程数大于等于最小线程数 则不需要创建
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        // 创建线程
        addWorker(null, false);
    }
}

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
      // 如果处于这三种情况不需要关闭线程池
      // 1. Running 状态
      // 2. SHUTDOWN 状态并且任务队列不为空，不能终止
      // 3. TIDYING 或者 TERMINATE 状态，说明已经在关闭了 不需要重复关闭
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 进入到关闭线程池的代码，如果线程池中还有线程，则需要打断线程
        if (workerCountOf(c) != 0) { // Eligible to terminate 可以关闭池子
            // 打断闲置线程，只打断一个
            interruptIdleWorkers(ONLY_ONE);
            return;
            // 如果有两个以上怎么办？只打断一个？
            // 这里只打断一个是因为 worker 回收的时候都会进入到该方法中来，可以回去再看看
            // runWorker方法最后的代码
        }
 
        // 线程已经回收完毕，准备关闭线程池
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();// 加锁
        try {
            // 将状态改变为 TIDYING 并且即将调用 terminated
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated(); // 终止线程池
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0)); // 改变状态
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
            // 如果终止失败会重试
        }
        // else retry on failed CAS
    }
}

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();//加锁～
    try {
      // 遍历线程池中的 wroker
        for (Worker w : workers) {
            Thread t = w.thread;
          // 如果线程没有被中断，并且能够获取到 worker的锁(说明是闲置线程)
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();// 中断线程
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
          // 只中断一个 worker 跳出循环，否则会将所有的闲置线程都中断
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();// 释放锁
    }

/**
 * Method invoked when the Executor has terminated. Default
 * implementation does nothing. Note: To properly nest multiple
 * overridings, subclasses should generally invoke
 * {@code super.terminated} within this method.
 */
protected void terminated() { }

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

void rejectedExecution(Runnable r, ThreadPoolExecutor executor);

private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();
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());
    }
}

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        r.run();
    }
}

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        e.getQueue().poll();
        e.execute(r);
    }
}