jdk的线程池实现-ThreadPoolExecutor

public interface Executor {
    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;
}

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

public static class DiscardPolicy implements RejectedExecutionHandler
public static class CallerRunsPolicy implements RejectedExecutionHandler
private final class Worker extends AbstractQueuedSynchronizer implements Runnable
public static class AbortPolicy implements RejectedExecutionHandler
public static class DiscardOldestPolicy implements RejectedExecutionHandler

AtomicInteger ctl; // 状态计数器
int RUNNING; // 运行状态
int SHUTDOWN ;// 关闭状态
int STOP; // 停止状态
int TIDYING; // 整理状态
int TERMINATED; //结束状态

static int runStateOf(int c) ; // 计算线程池当前状态
static boolean runStateLessThan(int c, int s) ; // 计算当前状态是否小于某个状态
static boolean runStateAtLeast(int c, int s) ; // 计算当前状态是否大于或等于某个状态
static boolean isRunning(int c) ; // 计算当前状态是否RUNNING

private volatile int corePoolSize;
private final BlockingQueue<Runnable> workQueue;
private volatile int maximumPoolSize;
private final ReentrantLock mainLock = new ReentrantLock();
private volatile RejectedExecutionHandler handler;
private volatile long keepAliveTime;

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

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) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            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;
    }

Runnable firstTask; // 表示新建的线程的第一个任务
boolean core; // 是否以核心线程数为边界，如果传入true，表示以核心线程数为边界，当前线程超过核心线程数则不创建新线程，如果不使用核心线程数为边界，则会以最大线程数为边界

int c = ctl.get();
int rs = runStateOf(c);

// Check if queue empty only if necessary.
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;
}

if (wc >= CAPACITY ||
		wc >= (core ? corePoolSize : maximumPoolSize))
			return false;

if (runStateOf(c) != rs)
		continue retry;

w = new Worker(firstTask);
final Thread t = w.thread;

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

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
	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();
}

mainLock.lock();

workers.add(w);
// ... 省略若干代码
workerAdded = true;

finally {
	mainLock.unlock();
}

if (workerAdded) {
	t.start();
	workerStarted = true;
}

finally {
	if (! workerStarted)
		addWorkerFailed(w);
}

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

public void run() {
	runWorker(this);
}

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

while (task != null || (task = getTask()) != null) 

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

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.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
}

void shutdown(); // 不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
List<Runnable> shutdownNow(); // 立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务

void setCorePoolSize(int corePoolSize);// 调整核心线程数
void setMaximumPoolSize(int maximumPoolSize) ; // 调整动态线程数

import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class Main {

    public static void main(String[] args) {
        ThreadPoolExecutor executor = (ThreadPoolExecutor)Executors.newFixedThreadPool(15);

        for(int i=0;i<15;i++){
            final int index = i;
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println("正在执行task "+index);
                    try {
                        Thread.currentThread().sleep(4000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("task "+index+"执行完毕");
                }
            });
            System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
                    executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
        }
        executor.shutdown();
    }
}