Java线程池原理

void execute(Runnable command);

public interface ExecutorService extends Executor {

    void shutdown();

    List<Runnable> shutdownNow();

    boolean isShutdown();

    boolean isTerminated();

    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> Future<T> submit(Callable<T> task);

    <T> Future<T> submit(Runnable task, T result);

    Future<?> submit(Runnable task);

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;            
}

public interface ThreadFactory {

    /**
     * Constructs a new {@code Thread}.  Implementations may also initialize
     * priority, name, daemon status, {@code ThreadGroup}, etc.
     *
     * @param r a runnable to be executed by new thread instance
     * @return constructed thread, or {@code null} if the request to
     *         create a thread is rejected
     */
    Thread newThread(Runnable r);
}

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.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

private static final int RUNNING    = -1 << COUNT_BITS;  // 10000000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;  // 00000000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;  // 00100000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;  // 01000000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;  // 01100000000000000000000000000000

* RUNNING -> SHUTDOWN
*    On invocation of shutdown(), perhaps implicitly in finalize()
* (RUNNING or SHUTDOWN) -> STOP
*    On invocation of shutdownNow()
* SHUTDOWN -> TIDYING
*    When both queue and pool are empty
* STOP -> TIDYING
*    When pool is empty
* TIDYING -> TERMINATED
*    When the terminated() hook method has completed
*
* Threads waiting in awaitTermination() will return when the
* state reaches TERMINATED.

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;

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 void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {
    // 这个循环将不断地从ctl变量中获取线程池的状态和线程数
    // 然后判断能否创建新的线程，如果可以，则更新ctl变量(线程数+1)
    // 更新失败就不断重试
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        // 这个if条件比较难理解，需要深入了解shutdown和shutdownNow方法
        // 和线程池的状态才能更好地理解这个if条件
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        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());

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        // 将线程池状态设置为SHUTDOWN
        advanceRunState(SHUTDOWN);
        // 中断空闲线程
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    // 尝试将线程池状态设置为TERMINATED
    tryTerminate();
}

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            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 interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 遍历所有工作线程，然后调用线程的interrupt方法中断线程
        for (Worker w : workers)
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}
void interruptIfStarted() {
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        try {
            t.interrupt();
        } catch (SecurityException ignore) {
        }
    }
}

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                	// 调用Thread的interrupt方法中断线程
                	// 实际上只是设置线程的中断标志为true
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            // 如果仅中断一个空闲线程，则在遍历完第一个工作线程的时候就会返回
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    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
            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;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // 因为如果任务不是正常完成的话，一般是发生了异常
    // 如果是线程是正常地被当作空闲线程中断的话，则会自动地减少线程数
    // 但是如果线程是在执行任务的过程中被中断的，
    // 应该手动调用decrementWorkerCount减少工作线程数
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    // 要访问线程安全的workers变量，需要加锁
    mainLock.lock();
    try {
    	// 统计该工作线程完成的任务数
        completedTaskCount += w.completedTasks;
        // 将当前工作线程从workers集合中移除
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    // 尝试将线程状态设置为TERMINATED
    // 因为线程终止可能是因为线程池被关闭了
    tryTerminate();

    int c = ctl.get();
    // 检测线程池的状态，如果状态为RUNNING或者SHUTDOWN
    // 则说明可能还有任务需要执行
    if (runStateLessThan(c, STOP)) {
    	// 如果线程是正常退出的，比如被当作空闲线程中断
    	// 则判断是否有必要添加一个工作线程取代它
        if (!completedAbruptly) {
        	// 如果核心线程运行超时，则最小线程数可以为0，否则至少为核心线程数
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            // 如果最小线程数可以为0，但是队列中还有任务，则至少需要一个工作线程来执行任务
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            // 如果当前工作线程数大于等于min，则没有必要添加工作线程取代退出的线程
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        // 如果线程是异常退出的，则一定会添加一个工作线程取代它
        addWorker(null, false);
    }
}

public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    // 将task包装成一个RunnableFuture
    // RunnableFuture继承了Runnable和Future接口
    RunnableFuture<T> ftask = newTaskFor(task, result);
    // 将调用ThreadPoolExecutor实现的execute方法
    execute(ftask);
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}

public void run() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

// FutureTask的其中一个构造器
public FutureTask(Runnable runnable, V result) {
    // 当传进来的是一个Runnable对象的时候，会调用Executors.callable方法
    // 将Runnable对象封装成Callable对象
    this.callable = Executors.callable(runnable, result);
    // FutureTask创建完成的初始状态为NEW
    this.state = NEW;       // ensure visibility of callable
}

// set方法，用于把任务执行结果保持到outcome变量中
protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    	// 这里把任务执行结果保持到了outcome变量
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}

// get方法，用于获取任务执行结果，如果任务没有执行完成会一直阻塞到任务执行完成
public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
    	// 等待任务执行完成，将一直阻塞到任务执行完成
        s = awaitDone(false, 0L);
    // 调用report方法获取结果
    return report(s);
}

// report方法，用于从outcome变量中取出结果
private V report(int s) throws ExecutionException {
    // 从outcome变量取出结果，outcome变量为Object类型
    Object x = outcome;
    if (s == NORMAL)
    	// 将结果转换成对象类型
        return (V)x;
    if (s >= CANCELLED)
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                	// 发送中断信号，如果任务能响应中断信号，则有可能会成功取消
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}

public class ThreadPoolTest {
    public static void main(String[] args) throws InterruptedException {
        // 新建一个线程池执行任务
        ThreadPoolExecutor exec = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(1));
        // 新建一个任务
        Runnable r = new Runnable() {
            public void run() {
            	// 死循环，每隔2秒输出一次task is running...
                while (true){
                    System.out.println("task is running...");
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                    	// 这里虽然能响应中断限号，但是接收到中断信号后没有退出循环
                    	// 所以任务将无法被取消，将会一直执行先去
                    	// 所以任务能不能够取消取决于任务的具体实现
                        System.out.println("interrupted!");
                    }
                }
            }
        };
        // 执行任务
        Future<StringBuffer> result = exec.submit(r, null);
        // 等待3秒，确保任务开始执行之后在取消任务
        // 如果任务还没有开始执行，是能够成功取消的
        // 因为任务开始之前会检测相应的状态位，状态为不对的话直接返回，不执行任务
        Thread.sleep(3000);
        // 取消正在执行的任务
        result.cancel(true);
        System.out.println("task was canceled.");
    }
}