012_线程池

2020-10-18   25 次阅读


一、线程池简介

1.0 使用规范

  1. 在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供,不允许在应用中自行显示的创建线程,这样一方面是线程的创建更加规范,可以合理控制开辟线程的数量;另一方面线程的细节管理交给线程池处理,优化了资源的开销。
  2. 而线程池不允许使用Executors去创建,而要通过ThreadPoolExecutor方式,这一方面是由于jdk中Executor框架虽然提供了如newFixedThreadPool()、newSingleThreadExecutor()、newCachedThreadPool()等创建线程池的方法,但都有其局限性,不够灵活;另外由于前面几种方法内部也是通过ThreadPoolExecutor方式实现,使用ThreadPoolExecutor有助于大家明确线程池的运行规则,创建符合自己的业务场景需要的线程池,避免资源耗尽的风险。

1.1 创建线程池

    public ThreadPoolExecutor(int corePoolSize, //线程池的核心线程数,没有任何任务时也会存在
                              int maximumPoolSize, //最大线程数,不管提交多少任务
                              long keepAliveTime, //线程的存活时间,没有等待到任务时线程会自动退出 
                              TimeUnit unit, //keepAliveTime的单位
                              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;

1.2 执行流程

image.png

   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);
    }
  1. 任务被提交到线程池,会先判断当前线程数量是否小于corePoolSize,如果小于则创建线程来执行提交的任务,否则将任务放入workQueue队列
  2. 如果workQueue满了,则判断当前线程数量是否小于maximumPoolSize,如果小于则创建线程执行任务,否则就会调用handler,以表示线程池拒绝接收任务。

1.3 自定义线程池

以下是自定义线程池,使用了有界队列,自定义ThreadFactory和拒绝策略的demo:

public class ThreadTest {

    public static void main(String[] args) throws InterruptedException, IOException {
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2);
        ThreadFactory threadFactory = new NameTreadFactory();
        RejectedExecutionHandler handler = new MyIgnorePolicy();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                workQueue, threadFactory, handler);
        executor.prestartAllCoreThreads(); // 预启动所有核心线程
        
        for (int i = 1; i <= 10; i++) {
            MyTask task = new MyTask(String.valueOf(i));
            executor.execute(task);
        }

        System.in.read(); //阻塞主线程
    }

    static class NameTreadFactory implements ThreadFactory {

        private final AtomicInteger mThreadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
            System.out.println(t.getName() + " has been created");
            return t;
        }
    }

    public static class MyIgnorePolicy implements RejectedExecutionHandler {

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            doLog(r, e);
        }

        private void doLog(Runnable r, ThreadPoolExecutor e) {
            // 可做日志记录等
            System.err.println( r.toString() + " rejected");
//          System.out.println("completedTaskCount: " + e.getCompletedTaskCount());
        }
    }

    static class MyTask implements Runnable {
        private String name;

        public MyTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                System.out.println(this.toString() + " is running!");
                Thread.sleep(3000); //让任务执行慢点
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "MyTask [name=" + name + "]";
        }
    }
}

image.png
其中线程线程1-4先占满了核心线程和最大线程数量,然后4、5线程进入等待队列,7-10线程被直接忽略拒绝执行,等1-4线程中有线程执行完后通知4、5线程继续执行。

1.4 预定义线程池

image.png
ExecutorService(ThreadPoolExecutor的顶层接口)使用线程池中的线程执行每个提交的任务,通常我们使用Executors的工厂方法来创建ExecutorService。

FixedThreadPool
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
  1. corePoolSize与maximumPoolSize相等,即其线程全为核心线程,是一个固定大小的线程池,是其优势;
  2. keepAliveTime = 0 该参数默认对核心线程无效,而FixedThreadPool全部为核心线程;
  3. workQueue 为LinkedBlockingQueue(无界阻塞队列),队列最大值为Integer.MAX_VALUE。如果任务提交速度持续大余任务处理速度,会造成队列大量阻塞。因为队列很大,很有可能在拒绝策略前,内存溢出。是其劣势;
  4. FixedThreadPool的任务执行是无序的;
  5. 适用场景:可用于Web服务瞬时削峰,但需注意长时间持续高峰情况造成的队列阻塞
CachedThreadPool
     public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
  1. corePoolSize = 0,maximumPoolSize = Integer.MAX_VALUE,即线程数量几乎无限制;
  2. keepAliveTime = 60s,线程空闲60s后自动结束。
  3. workQueue 为 SynchronousQueue 同步队列,这个队列类似于一个接力棒,入队出队必须同时传递,因为CachedThreadPool线程创建无限制,不会有队列等待,所以使用SynchronousQueue;
  4. 适用场景:快速处理大量耗时较短的任务,如Netty的NIO接受请求时,可使用
SingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

咋一瞅,不就是newFixedThreadPool(1)吗?定眼一看,这里多了一层FinalizableDelegatedExecutorService包装

    public static void main(String[] args) {
        ExecutorService fixedExecutorService = Executors.newFixedThreadPool(1);
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) fixedExecutorService;
        System.out.println(threadPoolExecutor.getMaximumPoolSize());
        threadPoolExecutor.setCorePoolSize(8);
        
        ExecutorService singleExecutorService = Executors.newSingleThreadExecutor();
//      运行时异常 java.lang.ClassCastException
//      ThreadPoolExecutor threadPoolExecutor2 = (ThreadPoolExecutor) singleExecutorService;
    }

对比可以看出,FixedThreadPool可以向下转型为ThreadPoolExecutor,并对其线程池进行配置,而SingleThreadExecutor被包装后,无法成功向下转型。因此,SingleThreadExecutor被定以后,无法修改,做到了真正的Single。

1.5 线程池状态

image.png

二、线程启动原理

Java多线程,皆始于Thread。Thread是多线程的根,每一个线程的开启都始于Thread的start()方法。那么线程是如何被开启,run方法是如何被执行的呢?
image.png

2.1 run与start

start()和run()方法的区别,为什么一定要用start()方法才是启动线程?
参看源码的解释:

 /**
     * Causes this thread to begin execution; the Java Virtual Machine
     * calls the <code>run</code> method of this thread.
     * 
     * 1、start方法将导致this thread开始执行。由JVM调用this thread的run方法。
     * 
     * The result is that two threads are running concurrently: the
     * current thread (which returns from the call to the
     * <code>start</code> method) and the other thread (which executes its
     * <code>run</code> method).
     * 
     * 2、结果是 调用start方法的当前线程 和 执行run方法的另一个线程 同时运行。
     * 
     * It is never legal to start a thread more than once.
     * In particular, a thread may not be restarted once it has completed
     * execution.
     *
     * 3、多次启动线程永远不合法。 特别是,线程一旦完成执行就不会重新启动。
     * 
     * @exception  IllegalThreadStateException  if the thread was already started.
     * 如果线程已启动,则抛出异常。
     * @see        #run()
     * @see        #stop()
     */
    public synchronized void start() {
        /**
         * This method is not invoked for the main method thread or "system"
         * group threads created/set up by the VM. Any new functionality added
         * to this method in the future may have to also be added to the VM.
         * 
         * 4、对于由VM创建/设置的main方法线程或“system”组线程,不会调用此方法。 
         *    未来添加到此方法的任何新功能可能也必须添加到VM中。
         * 
         * A zero status value corresponds to state "NEW".
         * 5、status=0 代表是 status 是 "NEW"。
         */
        if (threadStatus != 0)
            throw new IllegalThreadStateException();

        /* Notify the group that this thread is about to be started
         * so that it can be added to the group's list of threads
         * and the group's unstarted count can be decremented. 
         * 
         * 6、通知组该线程即将启动,以便将其添加到线程组的列表中,
         *    并且减少线程组的未启动线程数递减。
         * 
         * */
        group.add(this);

        boolean started = false;
        try {
            //7、调用native方法,底层开启异步线程,并调用run方法。
            start0();
            started = true;
        } finally {
            try {
                if (!started) {
                    group.threadStartFailed(this);
                }
            } catch (Throwable ignore) {
                /* do nothing. If start0 threw a Throwable then it will be passed up the call stack 
                 * 8、忽略异常。 如果start0抛出一个Throwable,它将被传递给调用堆栈。
                 */
            }
        }
    }

 //native方法,JVM创建并启动线程,并调用run方法
 private native void start0();
  1. start方法用synchronized修饰,为同步方法;
  2. 虽然为同步方法,但不能避免多次调用问题,用threadStatus来记录线程状态,如果线程被多次start会抛出异常;threadStatus的状态由JVM控制。
  3. 使用Runnable时,主线程无法捕获子线程中的异常状态。线程的异常,应在线程内部解决。

三、线程中断原理

3.1 暴力中断

优雅的中断线程,是一门艺术。Thread.stop, Thread.suspend, Thread.resume 都已经被废弃了。因为它们太暴力了,是不安全的,这种暴力中断线程是一种不安全的操作。

public class ThreadTest {
    public static void main(String[] args) throws InterruptedException {
        StopThread thread = new StopThread();
        thread.start();
        // 休眠1秒,确保线程进入运行
        Thread.sleep(1000);
        // 暂停线程
        thread.stop();
//      thread.interrupt();
        // 确保线程已经销毁
        while (thread.isAlive()) { }
        // 输出结果
        thread.print();
    }

    private static class StopThread extends Thread {

        private int x = 0; private int y = 0;

        @Override
        public void run() {
            // 这是一个同步原子操作
            synchronized (this) {
                ++x;
                try {
                    // 休眠3秒,模拟耗时操作
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                ++y;
            }
        }

        public void print() {
            System.out.println("x=" + x + " y=" + y);
        }
    }
}

上述代码中,run方法里是一个同步的原子操作,x和y必须要共同增加,然而这里如果调用thread.stop()方法强制中断线程,输出x=1 y=0。
没有异常,并破坏了我们的预期。如果这种问题出现在我们的程序中,会引发难以预期的异常。

3.2 interrupt()

述代码如果采用thread.interrupt()方法,输出结果如下:

x=1 y=1
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at ThreadTest$StopThread.run(ThreadTest.java:28)

x=1,y=1 这个结果是符合我们的预期,同时还抛出了个异常。

设计思想

interrupt() 它基于「一个线程不应该由其他线程来强制中断或停止,而是应该由线程自己自行停止。」思想,是一个比较温柔的做法,它更类似一个标志位。其实作用不是中断线程,而是「通知线程应该中断了」,具体到底中断还是继续运行,应该由被通知的线程自己处理。

一个线程如果有被中断的需求,那么就需要这样做:

  1. 在正常运行任务时,经常检查本线程的中断标志位,如果被设置了中断标志就自行停止线程。
  2. 在调用阻塞方法时正确处理InterruptedException异常。(例如:catch异常后就结束线程。)

3.3 interrupt相关方法

// 核心 interrupt 方法
   public void interrupt() {
        if (this != Thread.currentThread()) // 非本线程,需要检查权限
            checkAccess();

        synchronized (blockerLock) {
            Interruptible b = blocker;
            if (b != null) {
                interrupt0();           // 仅仅设置interrupt标志位
                b.interrupt(this);    // 调用如 I/O 操作定义的中断方法
                return;
            }
        }
        interrupt0();
    }
    // 静态方法,这个方法有点坑,调用该方法调用后会清除中断状态。
    public static boolean interrupted() {
        return currentThread().isInterrupted(true);
    }
    // 这个方法不会清除中断状态
    public boolean isInterrupted() {
        return isInterrupted(false);
    }
   // 上面两个方法会调用这个本地方法,参数代表是否清除中断状态
   private native boolean isInterrupted(boolean ClearInterrupted);
interrupt()
  1. interrupt 中断操作时,非自身打断需要先检测是否有中断权限,这由jvm的安全机制配置;
  2. 如果线程处于sleep, wait, join 等状态,那么线程将立即退出被阻塞状态,并抛出一个InterruptedException异常;
  3. 如果线程处于I/O阻塞状态,将会抛出ClosedByInterruptException(IOException的子类)异常;
  4. 如果线程在Selector上被阻塞,select方法将立即返回;
  5. 如果非以上情况,将直接标记 interrupt 状态;

注意:interrupt 操作不会打断所有阻塞,只有上述阻塞情况才在jvm的打断范围内,如处于锁阻塞的线程,不会受 interrupt 中断;

阻塞情况下中断

抛出异常后线程恢复非中断状态,即 interrupted = false

public class ThreadTest {
    
    public static void main(String[] args) throws InterruptedException {
        Thread t = new Thread(new Task("1"));
        t.start();
        t.interrupt();
    }

    static class Task implements Runnable{
        String name;
        
        public Task(String name) {
            this.name = name;
        }
        
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println("thread has been interrupt!");
            }
            System.out.println("isInterrupted: " + Thread.currentThread().isInterrupted());
            System.out.println("task " + name + " is over");
        }
    }
}
thread has been interrupt!
isInterrupted: false
task 1 is over
Thread.interrupted()

调用Thread.interrupted() 方法后线程恢复非中断状态

public class ThreadTest {
    
    public static void main(String[] args) throws InterruptedException {
        Thread t = new Thread(new Task("1"));
        t.start();
        t.interrupt();
    }

    static class Task implements Runnable{
        String name;
        
        public Task(String name) {
            this.name = name;
        }
        
        @Override
        public void run() {
            System.out.println("first :" + Thread.interrupted());
            System.out.println("second:" + Thread.interrupted());
            System.out.println("task " + name + " is over");
        }
    }
}
first :true
second:false
task 1 is over

上述两种隐含的状态恢复操作,是符合常理的,因为线程标记为中断后,用户没有真正中断线程,必然将其恢复为false。理论上Thread.interrupted()调用后,如果已中断,应该执行退出操作,不会重复调用。

四、多线程实现方式

本质上来讲只有一种方式:实现Runnable接口。
image.png

4.1 实现Runnable接口

public class DemoThreadTask implements Runnable{
    @Override
    public void run() {
        // TODO Auto-generated method stub
    }
    
    public static void main(String[] args) {
        DemoThreadTask task = new DemoThreadTask();
        Thread t = new Thread(task);
        t.start();
        ...
    }
}

实现Runnable接口,利用Runnable实例构造Thread,是较常用且最本质实现。此构造方法相当于对Runnable实例进行一层包装,在线程t启动时,调用Thread的run方法从而间接调用target.run()

4.2 继承Thread类

Thread类

public class Thread implements Runnable {
    /* What will be run. */
    private Runnable target;

    public void run() {
        if (target != null) {
            target.run();
        }
   }
     ...
}

继承Thread类

public class DemoThread extends Thread{
    @Override 
    //重写run方法
    public void run() {
        // TODO Auto-generated method stub
    }

    public static void main(String[] args) {
        DemoThread t = new DemoThread();
        t.start();
        ...
    }
}

需要注意的是继承Thread方式,target对象为null,重写了run方法,导致Thread原生的run方法失效,因此并不会调用到target.run()的逻辑,而是直接调用子类重写的run方法。

因为java是单根继承,此方式一般不常用。

4.3 实现Callable接口

public class DemoCallable implements Callable<String>{
    @Override
    public String call() throws Exception {
        // TODO Auto-generated method stub
        return null;
    }
    
    public static void main(String[] args) throws Exception {
        DemoCallable c = new DemoCallable();
        FutureTask<String> future = new FutureTask<>(c); 
        Thread t = new Thread(future);
        t.start();
        ...
        String result = future.get(); //同步获取返回结果
        System.out.println(result);
    }
}

实现Callable接口并通过FutureTask包装,可以获取到线程的处理结果,future.get()方法获取返回值,如果线程还没执行完,则会阻塞。

回看开篇的类图,FutureTask实现了RunnableFuture,RunnableFuture则实现了Runnable和Future两个接口。因此构造Thread时,FutureTask还是被转型为Runnable使用。因此其本质还是实现Runnable接口。

4.4 匿名内部类

匿名内部类也有多种变体,上述三种方式都可以使用匿名内部类来隐式实例化。

public class Demo{
    
    public static void main(String[] args) throws Exception {
        //方式一:Thread匿名内部类
        new Thread(){
            @Override
            public void run() {
                // TODO Auto-generated method stub
            }
        }.start();
        
        //方式二:Runnable匿名内部类
        new Thread(new Runnable() {
            @Override
            public void run() {
                // TODO Auto-generated method stub
            }
        }).start();
        
        ...
    }
}

4.5 Lambda表达式

Lambda表达式是jdk8引入的,已不是什么新东西,现在都jdk10了。demo如下:

public class Demo{
    public static void main(String[] args) throws Exception {
        new Thread(() -> System.out.println("running") ).start() ;
        ...
    }
}

4.6 线程池

public class DemoThreadTask implements Runnable{
    @Override
    public void run() {
        // TODO Auto-generated method stub
        System.out.println("running");
    }
    
    public static void main(String[] args) {
        DemoThreadTask task = new DemoThreadTask();
        ExecutorService ex = Executors.newCachedThreadPool();
        ex.execute(task);
        ...
    }
}

线程池与前面所述其他方式的区别在于执行线程的时候由ExecutorService去执行,最终还是利用Thread创建线程。
线程池的优势在于线程的复用,从而提高效率。

4.7 定时器

public class DemoTimmerTask {

    public static void main(String[] args) throws Exception {
        Timer timer = new Timer();
        timer.scheduleAtFixedRate((new TimerTask() {
            @Override
            public void run() {
                System.out.println("定时任务1执行了....");
            }
        }), 2000, 1000);
    }
}

TimerTask的实现了Runnable接口,Timer内部有个TimerThread继承自Thread,因此绕回来还是Thread + Runnable。

五、FutureTask实现原理

image.png

5.1 构造方法

    //构造方法一
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
    //构造方法二
    public FutureTask(Runnable runnable, V result) {
        //将runnable包装成callable
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

第一个构造方法我们比较熟悉,第二个构造方法可以用 Runnable 构造 FutureTask,将 Runnable 使用适配器模式 构造成 FutureTask ,使其具有 FutureTask 的特性,如可在主线程捕获Runnable的子线程异常。

5.2 run()方法

构造完FutureTask,就可以用FutureTask构造Thread,并启动线程。启动线程会调用FutureTask的run()方法,run()方法是FutureTask的实现关键

   public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    //1、获取返回值
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    //2、FutureTask的异常处理关键
                    setException(ex);
                }
                if (ran)
                    //3、设置返回值
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }
    //异常处理
     protected void setException(Throwable t) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            //设置为异常状态
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
            finishCompletion();
        }
    }
     //设置正常返回值
     protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            //设置为正常结束状态
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

在run()方法中,会调用callable对象的call()方法,并获取方法返回值,同时对call()方法中的异常进行了处理。异常时会将outcome设置为抛出的异常,正常时会将outcome设置为正常返回值,并将state设置成相应状态。

5.3 future.get()

future.get()获取线程返回结果

   public V get() throws InterruptedException, ExecutionException {
        int s = state;
        //未完成,则进入阻塞状态,等待完成
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);  //判断处理返回值
    }

    private V report(int s) throws ExecutionException {
        Object x = outcome;
        //根据state判断线程处理状态,并对outcome返回结果进行强转。
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x); //在主线程中抛出异常
    }

5.4 线程状态

//FutureTask定义的7种线程状态
    private volatile int state;
    private static final int NEW          = 0;
    private static final int COMPLETING   = 1; //设置返回值的过程,这个状态很短,可以划分为已完成状态。参考isDone()方法;
    private static final int NORMAL       = 2;
    private static final int EXCEPTIONAL  = 3;
    private static final int CANCELLED    = 4;
    private static final int INTERRUPTING = 5;
    private static final int INTERRUPTED  = 6;
    //是否已取消
    public boolean isCancelled() {
        return state >= CANCELLED;
    }
    //是否已完成
    public boolean isDone() {
        return state != NEW;
    }

线程的状态在执行过程不同阶段不断变化,这是FutureTask的状态控制关键。注意state是volatile修饰,保障了多线程间的可见性。

阻塞等待

线程status为NEW和COMPLETING的时候,会进入awaitDone方法,表示要等待完成。

   private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
            //线程是否被打断
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }

            int s = state;
            //已完成
            if (s > COMPLETING) {
                if (q != null)
                    q.thread = null;
                return s;
            } //正在处理返回值,这里时间很短,所以调用Thread.yield()方法,短时间的线程让步。
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
            else if (q == null) //创建等待节点
                q = new WaitNode();
            else if (!queued) //CAS把该线程加入等待队列
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            else if (timed) { //超时等待
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                //阻塞一段时间
                LockSupport.parkNanos(this, nanos);
            }
            else
                //线程阻塞,等待被唤醒
                LockSupport.park(this);
        }
    }
唤醒

在任务被取消、正常完成或执行异常时会调用finishCompletion()方法,从而唤醒等待队列中的线程。
image.png

private void finishCompletion() {
        // assert state > COMPLETING;
        for (WaitNode q; (q = waiters) != null;) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (;;) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        LockSupport.unpark(t); //唤醒线程
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }

        done();

        callable = null;        // to reduce footprint
    }

Q.E.D.

知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议