小菜鸟

// 最顶层接口：定义任务执行入口
public interface Executor {
    void execute(Runnable command);
}

// 扩展 Executor，增加线程池生命周期管理和任务提交能力
public interface ExecutorService extends Executor {
    void shutdown(); // 优雅关闭
    List<Runnable> shutdownNow(); // 强制关闭
    <T> Future<T> submit(Callable<T> task); // 提交有返回值的任务
    // 其他方法...
}

// 线程池的具体实现类
public class ThreadPoolExecutor extends AbstractExecutorService {
    // 核心实现...
}

  public ThreadPoolExecutor(int corePoolSize, // 核心线程数，决定新提交的任务是新开线程去执行还是放到任务队列中，当线程数量小于corePoolSize时才会去创建线程，如果大于corePoolSize会将任务放入workQueue队列中
                          int maximumPoolSize, // 最大线程数，线程池能创建的最大线程数，达到该数值后将不会创建新的线程，再添加任务则采用拒绝策略
                          long keepAliveTime, // 线程超过corePoolSize后，多余线程的最大闲置时间，如果超过，则会终止，但是对于核心线程，如果allowCoreThreadTimeOut(boolean value)设置为true的话，核心线程也会被终止
                          TimeUnit unit, // 时间单位
                          BlockingQueue<Runnable> workQueue, // 线程数如果大于corePoolSize会将任务放入workQueue队列中，存放未处理的任务的阻塞队列,常用的队列有
//1）ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小
//2）LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE
//3）SynchronousQueue：这个队列比较特殊，它不会进行存储，而是将直接新建一个线程来执行新来的任务
//4）PriorityBlockingQuene：具有优先级的无界阻塞队列
                          ThreadFactory threadFactory, // 生产线程的工厂类，可以用来定义线程名称以及线程的优先级
                          RejectedExecutionHandler handler)//拒绝策略

// 变量的高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 CAPACITY   = (1 << COUNT_BITS) - 1;

// 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

// 线程池状态  高3位
// 高三位111，表示运行中的状态标识，此时可以接受新任务并且执行队列中的任务 runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
// 高三位000，表示关闭中的状态标识，调用了shutdown方法，此时不再接收新任务，但是队列里的任务还得执行
private static final int SHUTDOWN   =  0 << COUNT_BITS;
// 高三位001，表示已停止的状态标识，调用了shutdownNow方法不再接受新任务，同时抛弃阻塞队列中的所有任务并中断正在执行的任务
private static final int STOP       =  1 << COUNT_BITS;
// 高三位010，表示当前所有任务已经终止，任务数量为0时的状态标识，在调用shutdown方法和shutdownNow方法时都会尝试更新这个状态
private static final int TIDYING    =  2 << COUNT_BITS;
// 高三位011，表示线程池已经完全终止（关闭），调用了terminated方法会更新为该状态
private static final int TERMINATED =  3 << COUNT_BITS;

// 获取运行状态  高3位
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 获取线程数量  低29位
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 计算ctl值 线程状态和线程个数 或操作
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 用来保存等待任务执行的阻塞队列
private final BlockingQueue<Runnable> workQueue;
// 可重入锁，对线程池状态等属性进行修改时需要持有该锁
private final ReentrantLock mainLock = new ReentrantLock();
// 包含了所有在工作的线程
private final HashSet<Worker> workers = new HashSet<Worker>();
// 锁条件队列，主要用于 awaitTermination  终止条件
private final Condition termination = mainLock.newCondition();
// 记录线程池最大工作线程的数量
private int largestPoolSize;
// 完成任务的数量，仅在中止工作任务时更新
private long completedTaskCount;
// 用于创建线程的工厂
private volatile ThreadFactory threadFactory;
// 拒绝策略
private volatile RejectedExecutionHandler handler;
// 空闲存活时间，如果线程池中的线程数量比核心线程数量还要多时，并且多出的这些线程都是闲置状态，该变量则是这些闲置状态的线程的存活时间
private volatile long keepAliveTime;
// 默认为 false，如果设为 true那么核心线程也会遵循 keepAliveTime的时间来做闲置处理
private volatile boolean allowCoreThreadTimeOut;
// 线程池核心线程数量
private volatile int corePoolSize;
// 线程池最大线程数量
private volatile int maximumPoolSize;

// submit是将任务进行一次封装，然后执行execute方法
public <T> Future<T> submit(Callable<T> task) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<T> ftask = newTaskFor(task);
  execute(ftask);
  return ftask;
}

public void execute(Runnable command) {
  		// command为空，则抛出空指针
        if (command == null)
            throw new NullPointerException();
        // 获取线程池状态 和 线程数量的组合值
        int c = ctl.get();
  			// workerCountOf方法根据ctl的低29位，得到线程池的当前线程数，如果线程数小于corePoolSize，则执行addWorker方法创建新的线程执行任务
  		// 判断是否创建核心线程
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true)) // 创建核心线程
              // 成功则直接返回
                return;
          // 失败需要再次获取线程池的状态
            c = ctl.get();
        }
  			// 经过上面的if语句，走到这里可能会因为两种情况1：当前工作线程数大于等于核心线程数 2：创建线程失败
  			// 判断线程池是否在运行
  			// 如果是RUNNING状态，则添加任务到阻塞队列中
        if (isRunning(c) && workQueue.offer(command)) {
          	// 双重检查，再次获取线程池状态，插入成功再次验证线程池是否运行，防止任务入队的过程中ctl的值发生变化
            int recheck = ctl.get();
          	//再次检查如果不在运行，移除已入队的任务，然后抛出拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
          	// 如果在运行，当前线程数为0，就启用一个非核心线程
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
  			// 加入阻塞队列失败(队列已满)，尝试创建非核心线程
  			//如果创建非核心线程失败，直接拒绝
        else if (!addWorker(command, false))
            reject(command);
    }


// 添加线程  core表示是否为核心线程
private boolean addWorker(Runnable firstTask, boolean core) {
  retry:
  for (;;) { // 自旋
    int c = ctl.get();
    // 当前线程池运行状态
    int rs = runStateOf(c);

    // Check if queue empty only if necessary.
    // 判断线程池状态，如果线程池的状态值大于或等于SHUTDOWN，则不处理提交任务，直接返回
    if (rs >= SHUTDOWN &&
        ! (rs == SHUTDOWN &&
           firstTask == null &&
           ! workQueue.isEmpty()))
      return false;
		// 自旋，更新线程数量
    for (;;) {
      // 当前线程数
      int wc = workerCountOf(c);
      // 根据core来判断是否为核心线程
      // 当前线程数已经超过容量，就直接返回
      if (wc >= CAPACITY ||
          wc >= (core ? corePoolSize : maximumPoolSize))
        return false;
      // CAS增加当前线程数量，更改成功结束自旋
      if (compareAndIncrementWorkerCount(c))
        break retry;
      // CAS操作失败了，获取最新值
      c = ctl.get();  // Re-read ctl
      // 检查线程池状态是不是修改了，线程池状态修改则需要重新获取线程池
      if (runStateOf(c) != rs)
        continue retry;
      // else CAS failed due to workerCount change; retry inner loop
    }
  }
// CAS操作成功，创建线程的所有条件都满足了，可以开始创建线程来执行任务
  // worker是否启动
  boolean workerStarted = false;
  // 是否将worker添加到workers集合中
  boolean workerAdded = false;
  Worker w = null;
  try {
    // 创建工作线程
    w = new Worker(firstTask);
    // 取出worker中的线程对象
    final Thread t = w.thread;
    if (t != null) {
      // 获取线程池主锁，为了保证workers同步
      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());

        //小于SHUTTDOWN即RUNNING
        //等于SHUTDOWN并且firstTask为null,不接受新的任务,但是会继续执行等待队列中的任务
        if (rs < SHUTDOWN ||
            (rs == SHUTDOWN && firstTask == null)) {
          // 判断线程是否是 alive状态
          if (t.isAlive()) // precheck that t is startable
            throw new IllegalThreadStateException();
          // 添加线程到workers中,workers是一个HashSet  private final HashSet<Worker> workers = new HashSet<Worker>();
          workers.add(w);
          int s = workers.size();
          if (s > largestPoolSize)
            largestPoolSize = s;
          // 设置当前工作者加入线程队列的已添加的标识为 true
          workerAdded = true;
        }
      } finally {
        mainLock.unlock();
      }
      // 线程添加线程池成功，启动新建的线程
      if (workerAdded) {
        t.start();
        workerStarted = true;
      }
    }
  } finally {
    // 线程启动失败
    if (! workerStarted)
      addWorkerFailed(w);
  }
  return workerStarted;
}

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
  
public void run() {
  runWorker(this);
}
// worker的执行方法
final void runWorker(Worker w) {
  		// 获取当前线程
        Thread wt = Thread.currentThread();
  		// 获取worker的firstTask
        Runnable task = w.firstTask;
        w.firstTask = null;
  		// 释放锁，设置aqs的state为0，允许中断
        w.unlock(); // allow interrupts
  		// 用于标识线程是否异常终止
        boolean completedAbruptly = true;
        try {
          // 循环getTask获取任务
            while (task != null || (task = getTask()) != null) {
              // 获取到可执行的任务，对worker对象加锁，保证线程在执行任务过程中不会被中断
                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) || // 线程池状态大于等于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 Runnable getTask() {
  // 标识是否超时
        boolean timedOut = false; // Did the last poll() time out?

        for (;;) {
          // 线程池控制状态
            int c = ctl.get();
          // 线程池运行状态
            int rs = runStateOf(c);

          // 如果线程池状态大于等于STOP，或者处于SHUTDOWN状态，并且阻塞队列为空，线程池工作线程数量递减，方法返回null，回收线程
            // 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?
           // 如果allowCoreThreadTimeOut为ture，或当前线程数大于核心池大小，则需要超时回收
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
					//如果worker数量大于maximumPoolSize或者超时了
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c)) //线程池工作线程数量递减，方法返回null，回收线程
                    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;
            }
        }
    }

// 取得第一个完成任务的结果值，当第一个任务执行完成后，会调用interrupt()将其他任务中断，需要结合if(Thread.currentThread().isInterrupted())来判断
// 会阻塞
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
    throws InterruptedException, ExecutionException 
// 等全部线程任务执行完毕后，取得所有完成任务的结果值
// 会阻塞
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException

// 示例：提交有返回值的任务
ExecutorService pool = new ThreadPoolExecutor(...);
Future<Integer> future = pool.submit(() -> {
    Thread.sleep(1000);
    return 1 + 1;
});
System.out.println(future.get()); // 阻塞等待结果：2

// 优雅关闭示例
pool.shutdown();
if (!pool.awaitTermination(1, TimeUnit.MINUTES)) {
    // 超时后强制关闭
    pool.shutdownNow();
}

// 示例：计算 1~100 的和（RecursiveTask）
public class SumTask extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 10; // 拆分阈值
    private int start;
    private int end;

    public SumTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        if (end - start <= THRESHOLD) {
            // 小任务直接计算
            int sum = 0;
            for (int i = start; i <= end; i++) sum += i;
            return sum;
        } else {
            // 大任务拆分
            int mid = (start + end) / 2;
            SumTask left = new SumTask(start, mid);
            SumTask right = new SumTask(mid + 1, end);
            left.fork(); // 并行执行左任务
            right.fork(); // 并行执行右任务
            return left.join() + right.join(); // 合并结果
        }
    }

    public static void main(String[] args) {
        ForkJoinPool pool = new ForkJoinPool();
        System.out.println(pool.invoke(new SumTask(1, 100))); // 输出：5050
    }
}