Java线程池-重点类源码解析--更新中

1.Runnable和Callable的区别

(1) Callable规定（重写）的方法是call()，Runnable规定（重写）的方法是run()

(2) Callable的任务执行后可返回值，而Runnable的任务是不能返回值的

(3) call方法可以抛出异常，run方法不可以

(4) 运行Callable任务可以拿到一个Future对象，表示异步计算的结果。它提供了检查计算是否完成的方法，以等待计算的完成，并检索计算的结果。通过Future对象可以了解任务执行情况，可取消任务的执行，还可获取执行结果

2. RunnableFuture接口和它的实现FutureTask

/**
 * A {@link Future} that is {@link Runnable}. Successful execution of
 * the {@code run} method causes completion of the {@code Future}
 * and allows access to its results.
 * @see FutureTask
 * @see Executor
 * @since 1.6
 * @author Doug Lea
 * @param <V> The result type returned by this Future's {@code get} method
 */
public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

我们可以看到Runnable继承了Runnable和Future，继承多接口肯定是为了实现功能的组合

FutureTask是它的具体实现，这个实现在线程池的很多地方有使用，比如AbstractExecutorService中的很多方法用到了newTaskFor(t)这个方法，方法声明如下：

/**
     * Returns a {@code RunnableFuture} for the given runnable and default
     * value.
     *
     * @param runnable the runnable task being wrapped
     * @param value the default value for the returned future
     * @param <T> the type of the given value
     * @return a {@code RunnableFuture} which, when run, will run the
     * underlying runnable and which, as a {@code Future}, will yield
     * the given value as its result and provide for cancellation of
     * the underlying task
     * @since 1.6
     */
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }

    /**
     * Returns a {@code RunnableFuture} for the given callable task.
     *
     * @param callable the callable task being wrapped
     * @param <T> the type of the callable's result
     * @return a {@code RunnableFuture} which, when run, will call the
     * underlying callable and which, as a {@code Future}, will yield
     * the callable's result as its result and provide for
     * cancellation of the underlying task
     * @since 1.6
     */
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

Future是什么？ Dog Lee给的定义如下：A Future represents the result of an asynchronous computation. 翻译过来就是代表异步执行的结果

我们可以理解为它是一个Stub，也就是我们扔到一个线程中执行的执行体的代理对象，通过它可以堆我们放进Runnable的对象进行观察以及取消等操作，主要方法如下：

AbstractExecutorService类的invokeAll方法

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
//常规的非空判断，空则抛出异常        
if (tasks == null)
            throw new NullPointerException();
        //根据Task的数量建立等量的Future的List
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        //标识是否已完成执行，这里我理解是全部执行完成
        boolean done = false;
        try {
            for (Callable<T> t : tasks) {
                //循环每个task，这里的创建的是一个FutureTask的对象
                RunnableFuture<T> f = newTaskFor(t);
                //结果List中添加这个FutureTask
                futures.add(f);
                //执行FutureTask
                execute(f);
            }
            for (int i = 0, size = futures.size(); i < size; i++) {
                //遍历FutureTask的list获取每个FutureTask
                Future<T> f = futures.get(i);
                //如果没有执行完成则阻塞等待
                if (!f.isDone()) {
                    try {
                        //为了阻塞等待，这个get的结果我们并不使用
                        f.get();
                    //需要忽略的两种异常，如果是其他异常就到finally中，也就是done=true和return不执行
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    }
                }
            }
            //全部执行完成且没有异常才会设置done为true并且返回future
            done = true;
            return futures;
        } finally {
            //如果最终没有执行完成则把没有执行的取消掉
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }





public boolean cancel(boolean mayInterruptIfRunning) {
        //如果线程的状态不为new则直接return false，森马也不执行
        //如果是new的话，尝试把new状态改为INTERRUPTING或者CANCELLED，具体根据mayInterruptIfRunning
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {//如果mayInterruptIfRunning是true，则执行以下操作
                try {
                    Thread t = runner;
                    //中断线程执行
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            finishCompletion();
        }
        return true;
    }

另外还有一个带超时时间的invokeAll

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            //这个for循环只是加进去，而不是直接执行，这里是为了不影响超时时间，添加也会消耗时间
            for (Callable<T> t : tasks)
                futures.add(newTaskFor(t));
            //根据当前时间计算超时的时候的时间
            final long deadline = System.nanoTime() + nanos;
            final int size = futures.size();

            // Interleave time checks and calls to execute in case
            // executor doesn't have any/much parallelism.
            // 这里才是开始真正的执行
            for (int i = 0; i < size; i++) {
                execute((Runnable)futures.get(i));
                //每一次循环都计算以下超时时间-当前时间
                nanos = deadline - System.nanoTime();
                //如果超时了，直接返回futures（包含未执行的），剩下的不再执行  
                if (nanos <= 0L)
                    return futures;
            }

            for (int i = 0; i < size; i++) {
                Future<T> f = futures.get(i);
                if (!f.isDone()) {
                    //监控执行过程的时候，如果超时了，直接返回不再执行
                    if (nanos <= 0L)
                        return futures;
                    try {
                        f.get(nanos, TimeUnit.NANOSECONDS);
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    } catch (TimeoutException toe) {
                        return futures;
                    }
                    //每次循环都算
                    nanos = deadline - System.nanoTime();
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }

InvokeAny方法

public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
        try {
            return doInvokeAny(tasks, false, 0);
        //没有设置超时时间，所以这里不会发生这种异常
        } catch (TimeoutException cannotHappen) {
            assert false;
            return null;
        }
    }

    //和不带超时时间的方法一样都是调用doInvokeAny方法
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return doInvokeAny(tasks, true, unit.toNanos(timeout));
    }


/**
     * invokeAny的具体的执行逻辑
     */
    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                              boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (tasks == null)
            throw new NullPointerException();
        int ntasks = tasks.size();
        if (ntasks == 0)
            throw new IllegalArgumentException();
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
        //注意这里的参数this代表使用的是当前的线程池，说明ECS持有的是当前的线程池
        ExecutorCompletionService<T> ecs =
            new ExecutorCompletionService<T>(this);

        // For efficiency, especially in executors with limited
        // parallelism, check to see if previously submitted tasks are
        // done before submitting more of them. This interleaving
        // plus the exception mechanics account for messiness of main
        // loop.

        try {
            // Record exceptions so that if we fail to obtain any
            // result, we can throw the last exception we got.
            ExecutionException ee = null;
            //根据是否有超时时间计算deadline
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            //取到tasks的迭代器
            Iterator<? extends Callable<T>> it = tasks.iterator();
            
            // Start one task for sure; the rest incrementally
            //ECS提交并返回存根加到futures
            futures.add(ecs.submit(it.next()));
            //提交一个减1
            --ntasks;
            //活跃数变成1
            int active = 1;

            for (;;) {
                //poll一般用于队列轮询，拿到当前所执行的task的执行结果
                Future<T> f = ecs.poll();
                //null代表没有执行完
                if (f == null) {
                    //ntasks最初代表的是任务的多少，大于0代表还没有执行完所有的
                    if (ntasks > 0) {
                        --ntasks;
                        //放入下一个任务
                        futures.add(ecs.submit(it.next()));
                        //活跃数加1
                        ++active;
                    }
                    //所有的活跃任务都执行完了
                    else if (active == 0)
                        break;
                    else if (timed) {
                        //考虑超时时间的情况
                        f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                        if (f == null)
                            throw new TimeoutException();
                        nanos = deadline - System.nanoTime();
                    }
                    //到这个else说明ntasks已经等于0了
                    //active不为0，也就是还没有全部执行完
                    //没有超时
                    else
                        //调用ecs的take方法，最终调用的是LinkedBlockingQueue的take方法
                        //会无限期等待，参考后面我贴的LinkedBlockingQueue的take方法
                        f = ecs.take();
                }
                //这里如果不是李大爷写的就是写的有问题，应该用else代替，但是老人家写的，这里不用else就是为了便于理解
                
                if (f != null) {
                    --active;
                    try {
                        return f.get();
                    } catch (ExecutionException eex) {
                        ee = eex;
                    } catch (RuntimeException rex) {
                        ee = new ExecutionException(rex);
                    }
                }
            }

            if (ee == null)
                ee = new ExecutionException();
            throw ee;

        } finally {
            for (int i = 0, size = futures.size(); i < size; i++)
                futures.get(i).cancel(true);
        }
    }

//以下两个方法来自ECS类(ExecutorCompletionService)
public Future<V> poll() {
        return completionQueue.poll();
}
//submit其实就是给QueueingFuture一个执行机会，执行完之后执行done方法的时候添加到completionQueue 里
public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        //执行的并不是传入的task，而是封装成了QueueingFuture
        //这里是装饰器模式
        executor.execute(new QueueingFuture(f));
        return f;
    }

    public Future<V> submit(Runnable task, V result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task, result);
        executor.execute(new QueueingFuture(f));
        return f;
    }

 public ExecutorCompletionService(Executor executor) {
        if (executor == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
            (AbstractExecutorService) executor : null;
        this.completionQueue = new LinkedBlockingQueue<Future<V>>();
    }



  private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }
        //
        //任务处理完成之后把当前放入completionQueue，自己把自己放进去
        //这个done方法由finishCompletion进行回调
        //FutureTask的cancel和set都可能调用finishCompletion
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }

public Future<V> take() throws InterruptedException {
        return completionQueue.take();
    }


//LinkedBlockingQueue的take方法
public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

invokeAny的总结：

不管有多少个任务，参照上面的doInvokeAny里面的f!=null这一段的内容

if (f != null) {
                    --active;
                    try {
                        return f.get();
                    } catch (ExecutionException eex) {
                        ee = eex;
                    } catch (RuntimeException rex) {
                        ee = new ExecutionException(rex);
                    }
                }

只要有一个线程执行成功了就直接返回作为这个方法的返回结果，然后取消剩下的任务的执行（实际只对NEW状态有作用，参照上面分析代码中的cancel方法）

小技巧：throw InterruptedException的一般为阻塞的

ECS的作用总结：（1）代理了线程池去执行（2）包装了futureTask，让任务自己执行完成后把自己放到completionQueue里，这个阻塞队列提供给外面的invokeAny方法来判断任务是否已执行完毕

poll方法一般是非阻塞的

上面是对invokeAny做的一个简单的测试我们看到因为第二个睡眠了然后造成被打断的异常，原因是第一个任务被执行完了，finally里会调用cancel方法把它唤醒，唤醒之后就输出了异常，如果我们不让他睡眠就不会出现这种情况

1.Executor：线程池顶级接口；

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null  
     */
    void execute(Runnable command);
}

定义了一个执行无返回值任务的方法，这里只是一个顶级的抽象，并不知道怎么去执行，具体执行的时候需要根据Executor的实现判断，可能再新线程、线程池、线程调用中执行

2.ExecutorService：线程池次级接口，对Executor做了一些扩展，增加了一些功能（线程池的服务）；

public interface ExecutorService extends Executor {

// 关闭线程池，不再接受新任务，但已经提交的任务会执行完成
void shutdown();
/**
* 立即关闭线程池，尝试停止正在运行的任务，未执行的任务将不再执行
* 被迫停止及未执行的任务将以列表的形式返回
*/
List<Runnable> shutdownNow();

// 检查线程池是否已关闭
boolean isShutdown();

// 检查线程池是否已终止，只有在shutdown()或shutdownNow()之后调用才有可能为true
boolean isTerminated();

// 在指定时间内线程池达到终止状态了才会返回true
boolean awaitTermination ( long timeout, TimeUnit unit) throws InterruptedException;

// 执行有返回值的任务，任务的返回值为task.call()的结果
<T> Future<T> submit ( Callable<T> task);

/**
   * 执行有返回值的任务，任务的返回值为这里传入的result
   * 当然只有当任务执行完成了调用get()时才会返回
   */
<T> Future<T> submit ( Runnable task, T result);

/**
   * 执行有返回值的任务，任务的返回值为null
   * 当然只有当任务执行完成了调用get()时才会返回
   */
Future<?> submit (Runnable task);

// 批量执行任务，只有当这些任务都完成了这个方法才会返回
<T> List<Future<T>> invokeAll ( Collection<? extends Callable<T>> tasks) throws InterruptedException;

/**
* 在指定时间内批量执行任务，未执行完成的任务将被取消
* 这里的timeout是所有任务的总时间，不是单个任务的时间
*/

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

3.ScheduledExecutorService：对ExecutorService做了一些扩展，增加一些定时任务相关的功能；

4.AbstractExecutorService：抽象类，运用模板方法设计模式实现了一部分方法；

5.ThreadPoolExecutor：普通线程池类，包含最基本的一些线程池操作相关的方法实现；

这个类的核心构造方法，这个是面试重点：

/**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    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;
    }

重点关注的参数：

（1） corePoolSize 核心线程数，什么时候都在（相当于固定工人，由没有活都会占用资源）

（2）maximumPoolSize最大线程数

maximumPoolSize - corePoolSize = 非核心线程数（相当于临时工）

（3）keepAliveTime 非核心线程数的闲置的最长时间（临时工回家前在公司呆一会，这个过程中可能由活可干）

（4）unit时间的单位

（5）workQueue 任务队列

（6）threadFactory 线程工厂，一般用默认的

默认线程工厂的实现（Executors类里）

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

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                //默认为false
                t.setDaemon(false);
            //优先级默认优先级
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

（7）handler拒绝策略

拒绝策略提供顶级接口 RejectedExecutionHandler ，其中方法 rejectedExecution 即定制具体的拒绝策略的执行逻辑。
jdk默认提供了四种拒绝策略：

AbortPolicy - 抛出异常，中止任务。抛出拒绝执行 RejectedExecutionException 异常信息。线程池默认的拒绝策略。必须处理好抛出的异常，否则会打断当前的执行流程，影响后续的任务执行

    /**
     * A handler for rejected tasks that throws a
     * {@code RejectedExecutionException}.
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

CallerRunsPolicy - 使用调用线程执行任务。当触发拒绝策略，只要线程池没有关闭的话，则使用调用线程直接运行任务。一般并发比较小，性能要求不高，不允许失败。但是，由于调用者自己运行任务，如果任务提交速度过快，可能导致程序阻塞，性能效率上必然的损失较大（类似于业务部门的产品提出需求，但是这边人员都占满了，还有很多在排队中的，也满了，这个时候产品带着他们自己的研发把这个需求做了）
DiscardPolicy - 直接丢弃，其他啥都没有（产品经理提需求过来，这边开发也不说能做不能做，默默的忘记这个需求），基本不用
DiscardOldestPolicy - 丢弃队列最老任务，添加新任务。当触发拒绝策略，只要线程池没有关闭的话，丢弃阻塞队列 workQueue 中最老的一个任务，并将新任务加入

————————————————
版权声明：本文为CSDN博主「鱼跃鹰飞」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/Chang_Yafei/article/details/126676068

ThreadPool的execute方法：


 /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null
     */
    public void execute(Runnable command) {
        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) {
            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);
    }

ctl的定义：
 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    //Integer.SIZE=32, 我们得到COUNT_BITS=29
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
    //runState一共由5种情况，用二进制位表示的话是3位（3位最多代表8个，2位最多代表4个）
    //分别使用-1，0，1，2，3左移29位表示runState（32位数里低位29位补0）
    // runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;

解释CTL之前我们先聊以下左移和右移的操作