介绍 线程 和 线程池 在 Java 中的基本概念、创建方式、核心参数和底层源码

NEW:
Thread thread = new Thread(()->{

});
System.out.println(thread.getState());

RUNNABLE:
Thread thread = new Thread(()->{
    while (true) {

    }
});
thread.start();
System.out.println(thread.getState());

BLOCKED:
Object object = new Object();
Thread thread = new Thread(()->{
    synchronized (object) {

    }
});
synchronized (object) {
    thread.start();
    Thread.sleep(500);
    System.out.println(thread.getState());
}

WAITING:
Object object = new Object();
Thread thread = new Thread(() -> {
    synchronized (object) {
        try {
            object.wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});
thread.start();
Thread.sleep(500);
System.out.println(thread.getState());

TIMED_WAITING:
Thread thread = new Thread(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
});
thread.start();
Thread.sleep(500);
System.out.println(thread.getState());

TERMINATED:
Thread thread = new Thread(() -> {
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
});
thread.start();
Thread.sleep(1000);
System.out.println(thread.getState());

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

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

private static class FinalizableDelegatedExecutorService
            extends DelegatedExecutorService {
    FinalizableDelegatedExecutorService(ExecutorService executor) {
        super(executor);
    }
    @SuppressWarnings("deprecation")
    protected void finalize() {
        super.shutdown();
    }
}

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

public static ScheduledExecutorService newScheduledThreadPool(
            int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue(), threadFactory);
}

public static ExecutorService newWorkStealingPool() {
    return new ForkJoinPool
        (Runtime.getRuntime().availableProcessors(),
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

// 是个int类型的数值 表达了两个意思 1: 声明当前线程池的状态 2: 声明线程池中的线程数
// 高3位是线程池状态 低29位是线程池中线程个数
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 29 方便后面的位运算
private static final int COUNT_BITS = Integer.SIZE - 3;
// 通过位运算得出最大容量
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
// 线程池状态
private static final int RUNNING    = -1 << COUNT_BITS; // 111 代表线程池为RUNNING,代表正常接收任务
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000 代表线程池为SHUTDOWN,不接收新任务,但是内部还会处理阻塞队列中的任务,正在进行的任务也会正常处理
private static final int STOP       =  1 << COUNT_BITS; // 001 代表线程池为STOP,不接收新任务,也不会处理阻塞队列中的任务,同时会中断正在进行的任务
private static final int TIDYING    =  2 << COUNT_BITS; // 010 代表线程池为TIDYING,过渡的状态,代表当前线程池即将 Game Over
private static final int TERMINATED =  3 << COUNT_BITS; // 011 代表线程池TERMINATED,代表当前线程池已经 Game Over,要执行terminated()

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~COUNT_MASK; } // 得到线程池的状态
private static int workerCountOf(int c)  { return c & COUNT_MASK; } // 得到当前线程池的线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; } // 得到上面提到的 32 位 int类型的数值

public void execute(Runnable command) {
  // 提交的任务不能为 null
  if (command == null)
    throw new NullPointerException();
  /*
    * Proceed in 3 steps:
    *
    * 1. If fewer than corePoolSize threads are running, try to
    * start a new thread with the given command as its first
    * task.  The call to addWorker atomically checks runState and
    * workerCount, and so prevents false alarms that would add
    * threads when it shouldn't, by returning false.
    *
    * 2. If a task can be successfully queued, then we still need
    * to double-check whether we should have added a thread
    * (because existing ones died since last checking) or that
    * the pool shut down since entry into this method. So we
    * recheck state and if necessary roll back the enqueuing if
    * stopped, or start a new thread if there are none.
    *
    * 3. If we cannot queue task, then we try to add a new
    * thread.  If it fails, we know we are shut down or saturated
    * and so reject the task.
    */
  // 获取核心线程 ctl,用于后面的判断
  int c = ctl.get();
  // 如果工作线程个数小于核心线程数
  // 满足要求,添加核心工作线程
  if (workerCountOf(c) < corePoolSize) {
    // addWorker(任务,是核心线程吗)
    // addWorker返回 true,代表添加工作线程成功
    // addWorker返回 false,代表添加工作线程失败
    // addWorker中会基于线程池状态,以及工作线程个数做判断,查看能否添加工作线程
    if (addWorker(command, true))
        // 工作线程构建出来了,任务也交给 command 去处理了
        return;
    // 说明线程池状态或者是工作线程个数发生了变化,导致添加失败,重新获取一次 ctl
    c = ctl.get();
  }
  // 添加核心工作线程失败,调用一下
  // 判断线程池状态是否是 RUNNING,如果是,正常基于阻塞队列的 offer 方法,将任务添加到阻塞队列
  if (isRunning(c) && workQueue.offer(command)) {
    // 如果任务添加到阻塞队列成功,走 if 内部
    // 如果任务在扔到阻塞队列之前,线程池状态突然改变了
    // 重新获取 ctl
    int recheck = ctl.get();
    // 如果线程池的状态不是 RUNNING,将任务从阻塞队列中移除
    if (! isRunning(recheck) && remove(command))
      // 并且直接拒绝策略
      reject(command);
    // 在这,说明阻塞队列有我刚刚放进去的任务
    // 查看一下工作线程数是不是 0 个
    // 如果工作线程为 0 个,需要添加一个非核心工作线程去处理阻塞队列中的任务
    // 发生这种情况有两种:
    // 1.构建线程池时,核心线程数为 0 个
    // 2.即便有核心线程,可以设置核心线程也允许超时,设置 allowCoreThreadTimeOut 为 true,代表核心线程也可以关闭
    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();;) {
    // Check if queue empty only if necessary.
    // 检查线程池的运行状态是否 shutdown,任务队列是否为空等状态是否正常
    if (runStateAtLeast(c, SHUTDOWN)
      && (runStateAtLeast(c, STOP)
        || firstTask != null
        || workQueue.isEmpty()))
      return false;

    // 循环更新状态,和线程数量的更新,都是使用 CAS 的模式更新
    for (;;) {
      // 检查运行的线程数是否超标
      if (workerCountOf(c)
        >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
        return false;
      // CAS 方式更新线程数量
      if (compareAndIncrementWorkerCount(c))
        // 更新成功后直接跳转到方法第一行,并且不在进入这些 for 循环中
        // 因为状态啥的已经更新成功了
        break retry;
      c = ctl.get();  // Re-read ctl
      // 如果 CAS 增加线程数失败,检查下状态是否还是和之前一样
      if (runStateAtLeast(c, SHUTDOWN))
        // 不一样了,会跳转到第一行,并会重新进入 for 循环中走一遍流程
        continue retry;
      // else CAS failed due to workerCount change; retry inner loop
    }
  }

  boolean workerStarted = false;
  boolean workerAdded = false;
  Worker w = null;
  try {
    // 新建一个 worker,内部包含线程
    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 c = ctl.get();

        // 二次检查线程池的状态
        if (isRunning(c) ||
          (runStateLessThan(c, STOP) && firstTask == null)) {
          // 池的状态没有问题,检查线程是否存活
          if (t.getState() != Thread.State.NEW)
            throw new IllegalThreadStateException();
          // worker 放入池的 hashset 中
          workers.add(w);
          workerAdded = true;
          int s = workers.size();
          // 并将 worker 的数量赋予池中的 largestPoolSize 成员变量
          if (s > largestPoolSize)
            largestPoolSize = s;
        }
      } finally {
        mainLock.unlock();
      }
      // worker 创建成功并加入 workers 成功
      if (workerAdded) {
        // 启动线程
        t.start();
        // 修改状态
        workerStarted = true;
      }
    }
  } finally {
    if (! workerStarted)
      // 启动线程失败,进行失败处理
      addWorkerFailed(w);
  }
  return workerStarted;
}

// 如果可以在不超过队列容量的情况下立即插入指定的元素,则在该队列的尾部插入该元素;如果成功,则返回true;如果该队列已满,则返回false
public boolean offer(E e) {
    // 不允许元素为空
    Objects.requireNonNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock(); // 加锁 保证调用 offer 方法的时候只有一个线程
    try {
        // 队列满
        if (count == items.length)
            return false;
        else {
            // 队列未满
            enqueue(e);
            return true;
        }
    } finally {
        // 释放锁 让其他线程可以调用 offer 方法
        lock.unlock();
    }
}

// 在当前放置位置插入元素,前进和发出信号。只有在锁定的情况下才能呼叫。
private void enqueue(E e) {
    final Object[] items = this.items;
    // 元素添加到数组中
    items[putIndex] = e;
    // 当索引满了 修改为0
    if (++putIndex == items.length) putIndex = 0;
    count++;
    // 使用条件对象 notEmpty 通知 比如使用 take 方法的时候队列中没有数据 被阻塞 这个时候队列中插入了数据 需要调用 signal() 进行通知
    notEmpty.signal();
}

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable {
  /**
   * This class will never be serialized, but we provide a
   * serialVersionUID to suppress a javac warning.
   */
  private static final long serialVersionUID = 6138294804551838833L;

  /** Thread this worker is running in.  Null if factory fails. */
  final Thread thread;
  /** Initial task to run.  Possibly null. */
  Runnable firstTask;
  /** Per-thread task counter */
  volatile long completedTasks;

  // TODO: switch to AbstractQueuedLongSynchronizer and move
  // completedTasks into the lock word.

  /**
   * Creates with given first task and thread from ThreadFactory.
   * @param firstTask the first task (null if none)
   */
  Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    // 创建 worker 线程
    this.thread = getThreadFactory().newThread(this);
  }

  /** Delegates main run loop to outer runWorker. */
  public void run() {
    runWorker(this);
  }

  // Lock methods
  //
  // The value 0 represents the unlocked state.
  // The value 1 represents the locked state.

  private boolean isHeldExclusively() {
    return getState() != 0;
  }

  private boolean tryAcquire(int unused) {
    if (compareAndSetState(0, 1)) {
      setExclusiveOwnerThread(Thread.currentThread());
      return true;
    }
    return false;
  }

  private boolean tryRelease(int unused) {
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
  }

  public void lock() {
    acquire(1);
  }

  public boolean tryLock() {
    return tryAcquire(1);
  }

  public void unlock() {
    release(1);
  }

  public boolean isLocked() {
    return isHeldExclusively();
  }

  void interruptIfStarted() {
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
      try {
        t.interrupt();
      } catch (SecurityException ignore) {
      }
    }
  }
}

final void runWorker(Worker w) {
    // 获取当前线程
    Thread wt = Thread.currentThread();
    // 获取任务
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 任务不为空 执行任务 如果任务为空 通过getTask()从阻塞队列中获取任务
        while (task != null || (task = getTask()) != null) {
            // 加锁 避免被 SHUTDOWN 任务也不会中断
            w.lock();
            // 如果线程池状态大于等于STOP,请确保线程被中断;如果没有,请确保线程没有中断。这需要在第二种情况下重新检查,以处理关闭在清除中断时无竞争
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                // 中断
                wt.interrupt();
            try {
                // 执行任务前的操作
                beforeExecute(wt, task);
                try {
                    task.run();
                    afterExecute(task, null);
                } catch (Throwable ex) {
                    // 执行任务后的操作
                    afterExecute(task, ex);
                    throw ex;
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}