Java concurrency线程池原理 Java concurrency线程池之线程池原理(二)_动力节点Java学院整理

import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

public class ThreadPoolDemo1 {

  public static void main(String[] args) {
    // 创建一个可重用固定线程数的线程池
    ExecutorService pool = Executors.newFixedThreadPool(2);
    // 创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口
    Thread ta = new MyThread();
    Thread tb = new MyThread();
    Thread tc = new MyThread();
    Thread td = new MyThread();
    Thread te = new MyThread();
    // 将线程放入池中进行执行
    pool.execute(ta);
    pool.execute(tb);
    pool.execute(tc);
    pool.execute(td);
    pool.execute(te);
    // 关闭线程池
    pool.shutdown();
  }
}

class MyThread extends Thread {

  @Override
  public void run() {
    System.out.println(Thread.currentThread().getName()+ " is running.");
  }
}

pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.

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

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

public static ThreadFactory defaultThreadFactory() {
  return new DefaultThreadFactory();
}

 static class DefaultThreadFactory implements ThreadFactory {
  private static final AtomicInteger poolNumber = new AtomicInteger(1);
  private final ThreadGroup group;
  private final AtomicInteger threadNumber = new AtomicInteger(1);
  private final String namePrefix;

  DefaultThreadFactory() {
    SecurityManager s = System.getSecurityManager();
    group = (s != null) ? s.getThreadGroup() :
               Thread.currentThread().getThreadGroup();
    namePrefix = "pool-" +
           poolNumber.getAndIncrement() +
           "-thread-";
  }

  // 提供创建线程的API。
  public Thread newThread(Runnable r) {
    // 线程对应的任务是Runnable对象r
    Thread t = new Thread(group, r,
               namePrefix + threadNumber.getAndIncrement(),
               0);
    // 设为“非守护线程”
    if (t.isDaemon())
      t.setDaemon(false);
    // 将优先级设为“Thread.NORM_PRIORITY”
    if (t.getPriority() != Thread.NORM_PRIORITY)
      t.setPriority(Thread.NORM_PRIORITY);
    return t;
  }
}

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 execute(Runnable command) {
  // 如果任务为null，则抛出异常。
  if (command == null)
    throw new NullPointerException();
  // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
  int c = ctl.get();
  // 当线程池中的任务数量 < "核心池大小"时，即线程池中少于corePoolSize个任务。
  // 则通过addWorker(command, true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
  if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
      return;
    c = ctl.get();
  }
  // 当线程池中的任务数量 >= "核心池大小"时，
  // 而且，"线程池处于允许状态"时，则尝试将任务添加到阻塞队列中。
  if (isRunning(c) && workQueue.offer(command)) {
    // 再次确认“线程池状态”，若线程池异常终止了，则删除任务；然后通过reject()执行相应的拒绝策略的内容。
    int recheck = ctl.get();
    if (! isRunning(recheck) && remove(command))
      reject(command);
    // 否则，如果"线程池中任务数量"为0，则通过addWorker(null, false)尝试新建一个线程，新建线程对应的任务为null。
    else if (workerCountOf(recheck) == 0)
      addWorker(null, false);
  }
  // 通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
  // 如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内容。
  else if (!addWorker(command, false))
    reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {
  retry:
  // 更新"线程池状态和计数"标记，即更新ctl。
  for (;;) {
    // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
    int c = ctl.get();
    // 获取线程池状态。
    int rs = runStateOf(c);

    // 有效性检查
    if (rs >= SHUTDOWN &&
      ! (rs == SHUTDOWN &&
        firstTask == null &&
        ! workQueue.isEmpty()))
      return false;

    for (;;) {
      // 获取线程池中任务的数量。
      int wc = workerCountOf(c);
      // 如果"线程池中任务的数量"超过限制，则返回false。
      if (wc >= CAPACITY ||
        wc >= (core ? corePoolSize : maximumPoolSize))
        return false;
      // 通过CAS函数将c的值+1。操作失败的话，则退出循环。
      if (compareAndIncrementWorkerCount(c))
        break retry;
      c = ctl.get(); // Re-read ctl
      // 检查"线程池状态"，如果与之前的状态不同，则从retry重新开始。
      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 {
    final ReentrantLock mainLock = this.mainLock;
    // 新建Worker，并且指定firstTask为Worker的第一个任务。
    w = new Worker(firstTask);
    // 获取Worker对应的线程。
    final Thread t = w.thread;
    if (t != null) {
      // 获取锁
      mainLock.lock();
      try {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 再次确认"线程池状态"
        if (rs < SHUTDOWN ||
          (rs == SHUTDOWN && firstTask == null)) {
          if (t.isAlive()) // precheck that t is startable
            throw new IllegalThreadStateException();
          // 将Worker对象(w)添加到"线程池的Worker集合(workers)"中
          workers.add(w);
          // 更新largestPoolSize
          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 Future<?> submit(Runnable task) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<Void> ftask = newTaskFor(task, null);
  execute(ftask);
  return ftask;
}

public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  // 获取锁
  mainLock.lock();
  try {
    // 检查终止线程池的“线程”是否有权限。
    checkShutdownAccess();
    // 设置线程池的状态为关闭状态。
    advanceRunState(SHUTDOWN);
    // 中断线程池中空闲的线程。
    interruptIdleWorkers();
    // 钩子函数，在ThreadPoolExecutor中没有任何动作。
    onShutdown(); // hook for ScheduledThreadPoolExecutor
  } finally {
    // 释放锁
    mainLock.unlock();
  }
  // 尝试终止线程池
  tryTerminate();
}