钻石 01：明心见性 - 如何由表及里精通线程池设计与原理

/** * 王者线程池 */public class TheKingThreadPool {    private final BlockingQueue<Task> taskQueue;    private final List<Worker> workers = new ArrayList<>();    private ThreadPoolStatus status;
    /**     * 初始化构建线程池     *     * @param worksNumber 线程池中的工作线程数量     * @param taskQueue   任务队列     */    public TheKingThreadPool(int worksNumber, BlockingQueue<Task> taskQueue) {        this.taskQueue = taskQueue;        status = ThreadPoolStatus.RUNNING;        for (int i = 0; i < worksNumber; i++) {            workers.add(new Worker("Worker" + i, taskQueue));        }        for (Worker worker : workers) {            Thread workThread = new Thread(worker);            workThread.setName(worker.getName());            workThread.start();        }    }
    /**     * 提交任务     *     * @param task 待执行的任务     */    public synchronized void execute(Task task) {        if (!this.status.isRunning()) {            throw new IllegalStateException("线程池非运行状态，停止接单啦~");        }        this.taskQueue.offer(task);    }
    /**     * 等待所有任务执行结束     */    public synchronized void waitUntilAllTasksFinished() {        while (this.taskQueue.size() > 0) {            try {                Thread.sleep(1);            } catch (InterruptedException e) {                e.printStackTrace();            }        }    }
    /**     * 关闭线程池     */    public synchronized void shutdown() {        this.status = ThreadPoolStatus.SHUTDOWN;    }
    /**     * 停止线程池     */    public synchronized void stop() {        this.status = ThreadPoolStatus.SHUTDOWN;        for (Worker worker : workers) {            worker.doStop();        }    }}

/** * 线程池中用于执行任务的线程 */public class Worker implements Runnable {    private final String name;    private Thread thread = null;    private final BlockingQueue<Task> taskQueue;    private boolean isStopped = false;    private AtomicInteger counter = new AtomicInteger();
    public Worker(String name, BlockingQueue<Task> queue) {        this.name = name;        taskQueue = queue;    }
    public void run() {        this.thread = Thread.currentThread();        while (!isStopped()) {            try {                Task task = taskQueue.poll(5L, TimeUnit.SECONDS);                if (task != null) {                    note(this.thread.getName(), ":获取到新的任务->", task.getTaskDesc());                    task.run();                    counter.getAndIncrement();                }            } catch (Exception ignored) {            }        }        note(this.thread.getName(), ":已结束工作，执行任务数量：" + counter.get());    }
    public synchronized void doStop() {        isStopped = true;        if (thread != null) {            this.thread.interrupt();        }    }
    public synchronized boolean isStopped() {        return isStopped;    }
    public String getName() {        return name;    }}

/** * 任务 */public interface Task extends Runnable {    String getTaskDesc();}

public enum ThreadPoolStatus {    RUNNING(),    SHUTDOWN(),    STOP(),    TIDYING(),    TERMINATED();
    ThreadPoolStatus() {    }
    public boolean isRunning() {        return ThreadPoolStatus.RUNNING.equals(this);    }}

 public static void main(String[] args) {        TheKingThreadPool theKingThreadPool = new TheKingThreadPool(3, new ArrayBlockingQueue<>(10));
        String[] wildMonsters = {"棕熊", "野鸡", "灰狼", "野兔", "狐狸", "小鹿", "小花豹", "野猪"};        for (String wildMonsterName : wildMonsters) {            theKingThreadPool.execute(new Task() {                public String getTaskDesc() {                    return wildMonsterName;                }
                public void run() {                    System.out.println(Thread.currentThread().getName() + ":" + wildMonsterName + "已经烤好");                }            });        }
        theKingThreadPool.waitUntilAllTasksFinished();        theKingThreadPool.stop();    }

Worker0:获取到新的任务->灰狼Worker1:获取到新的任务->野鸡Worker1:野鸡已经烤好Worker2:获取到新的任务->棕熊Worker2:棕熊已经烤好Worker1:获取到新的任务->野兔Worker1:野兔已经烤好Worker0:灰狼已经烤好Worker1:获取到新的任务->小鹿Worker1:小鹿已经烤好Worker2:获取到新的任务->狐狸Worker2:狐狸已经烤好Worker1:获取到新的任务->野猪Worker1:野猪已经烤好Worker0:获取到新的任务->小花豹Worker0:小花豹已经烤好Worker0:已结束工作，执行任务数量：2Worker2:已结束工作，执行任务数量：2Worker1:已结束工作，执行任务数量：4
Process finished with exit code 0

// 线程池控制相关的主要变量// 这个变量很神奇，下文后专门陈述，请特别留意private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 待处理的任务队列private final BlockingQueue < Runnable > workQueue;// 工作线程集合private final HashSet < Worker > workers = new HashSet < Worker > ();// 创建线程所用到的线程工厂private volatile ThreadFactory threadFactory;// 拒绝策略private volatile RejectedExecutionHandler handler;// 核心线程数private volatile int corePoolSize;// 最大线程数private volatile int maximumPoolSize;// 空闲线程的保活时长private volatile long keepAliveTime;// 线程池变更的主要控制锁，在工作线程数、变更线程池状态等场景下都会用到private final ReentrantLock mainLock = new ReentrantLock();

/**    计算当前线程池的状态*/private static int runStateOf(int c) {    return c & ~CAPACITY;}/**    计算当前工作线程数*/private static int workerCountOf(int c) {    return c & CAPACITY;}/**    初始化ctl变量*/private static int ctlOf(int rs, int wc) {    return rs | wc;}

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

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

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

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

/*** 提交Runnable类型的任务并执行，但不返回结果*/public void execute(Runnable command){...}/*** 提交Runnable类型的任务，并返回结果*/public Future<?> submit(Runnable task){...}/*** 提交Runnable类型的任务，并返回结果，支持指定默认结果*/public <T> Future<T> submit(Runnable task, T result){...}/*** 提交Callable类型的任务并执行*/public <T> Future<T> submit(Callable<T> task) {...}/*** 关闭线程池，继续执行队列中未完成的任务，但不会接收新的任务*/public void shutdown() {...}/*** 立即关闭线程池，同时放弃未执行的任务，并不再接收新的任务*/public List<Runnable> shutdownNow(){...}

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

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

public BlockingQueue<Runnable> getQueue() {    return workQueue;}

public boolean remove(Runnable task) {    boolean removed = workQueue.remove(task);    tryTerminate(); // In case SHUTDOWN and now empty    return removed;}

  private boolean addWorker(Runnable firstTask, boolean core) {...}

protected void beforeExecute(Thread t, Runnable r) { }

protected void afterExecute(Runnable r, Throwable t) { }

protected void terminated() { }

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) {           ...        } finally {            processWorkerExit(w, completedAbruptly); // 主动回收自己        }    }

public static void main(String[] args) {    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 20, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue < > (10));
    String[] wildMonsters = {"棕熊", "野鸡", "灰狼", "野兔", "狐狸", "小鹿", "小花豹", "野猪"};    for (String wildMonsterName: wildMonsters) {        threadPoolExecutor.execute(new RunnableTask() {            public String getTaskDesc() {                return wildMonsterName;            }
            public void run() {                System.out.println(Thread.currentThread().getName() + ":" + wildMonsterName + "已经烤好");            }        });    }
    threadPoolExecutor.shutdown();}

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

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

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

 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)             if (w.isLocked())                 ++n;         return n;     } finally {         mainLock.unlock();     } }

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