当前位置 : 主页 > 编程语言 > java >

Java线程池submit阻塞获取结果的实现原理详解

来源:互联网 收集:自由互联 发布时间:2023-01-30
目录 前言 案例演示 和execute区别 原理实现 RunnableFuture类介绍 任务执行run()原理 任务结果获取get()原理 任务取消cancel()原理 总结 前言 Java线程池中提交任务运行,通常使用 execute() 方法
目录
  • 前言
  • 案例演示
  • 和execute区别
  • 原理实现
    • RunnableFuture类介绍
    • 任务执行run()原理
    • 任务结果获取get()原理
    • 任务取消cancel()原理
  • 总结

    前言

    Java线程池中提交任务运行,通常使用execute()方法就足够了。那如果想要实现在主线程中阻塞获取线程池任务运行的结果,该怎么办呢?答案是用submit()方法提交任务。这也是面试中经常被问到的一个知识点,execute()submit()提交任务的的区别是什么?底层是如何实现的?

    案例演示

    现在我们通过简单的例子演示下submit()方法的妙处。

    @Test
    public void testSubmit() throws ExecutionException, InterruptedException {
        // 创建一个核心线程数为5的线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(50));
    
        // 创建一个计算任务
        Callable<Integer> myTask = new Callable<Integer>() {
    
            @Override
            public Integer call() throws Exception {
                int result = 0;
                for (int i = 0; i < 10000; i++) {
                    result += i;
                }
                Thread.sleep(1000);
                return result;
            }
        };
    
        log.info("start submit task .....");
        Future<Integer> future = threadPoolExecutor.submit(myTask);
    
        Integer sum = future.get();
        log.info("get submit result: [{}]", sum);
    
        // use sum do other things
    }
    

    运行结果:

    主线程的确阻塞等待线程返回。

    Future类API

    我们看到用submit提交任务最后返回一个Future对象,Future表示异步计算的结果。那它都提供了什么API呢?

    方法说明V get()等待任务执行完成,然后获取其结果。V get(long timeout, TimeUnit unit)等待获取任务执行的结果,如果任务超过一定时间没有执行完毕,直接返回,抛出异常,不会一直等待下去。boolean isDone()如果此任务已完成,则返回true。完成可能是由于正常终止、异常或取消——在所有这些情况下,该方法都将返回true。boolean isCancelled()如果该任务在正常完成之前被取消,则返回true。boolean cancel(boolean mayInterruptIfRunning)试图取消此任务的执行。1. 如果任务已经完成、已经取消或由于其他原因无法取消,则此尝试将失败。
    • 如果在调用cancel时此任务尚未启动,则此任务不应运行。
    • 如果任务已经开始,那么mayInterruptIfRunning参数确定是否应该中断执行此任务的线程以试图停止该任务。 |

    和execute区别

    从功能层面,我们已经很明白他们最大区别,

    • execute()方式提交任务没有返回值,直接线程中池异步运行任务。
    • submit()方式提交任务有返回值Future, 调用get方法可以阻塞调用线程,等待任务运行返回的结果。

    那从源码层面,二者又有什么区别和联系呢?

    我们看下submit()提交的入口方法,代码如下:

    // AbstractExecutorService#submit
    public <T> Future<T> submit(Callable<T> task) {
        // 判空处理
        if (task == null) throw new NullPointerException();
        // 将提交的任务包装成RunnableFuture
        RunnableFuture<T> ftask = newTaskFor(task);
        // 最终还是调用execute方法执行任务
        execute(ftask);
        return ftask;
    }
    

    殊途同归,最终都是调用execute()方法,只不过submit()方法在调用前做一层包装,将任务包装成RunnableFuture对象。

    关于线程池中execute()方法提交的流程和原理实现不理解的,强烈建议先学习这篇文章:Java线程池源码深度解析。

    原理实现

    本节内容我们聚焦在submit()方法的实现原理。

    我们先思考下,如果让我们设计实现调用get阻塞知道线程返回结果,要考虑哪些方面呢?

    • 任务是否执行结束或者执行出错等情况,是不是需要有个状态位标记?
    • 任务的执行结果如何保存?
    • 如果任务没有执行结束,如何阻塞当前线程,LockSupport.park()是一种方式。
    • 如果有多个外部线程获取get,是不是应该也要把外部线程存下来,怎么存?因为后面任务执行完后需要唤醒他们。

    带着这些问题和基本思路我们看下jdk8中是如何实现的。

    RunnableFuture类介绍

    submit()方法中调用newTaskFor()方法获取RunnableFuture对象。

    // AbstractExecutorService#newTaskFor
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        // 调用FutureTask的构造方法返回RunnableFuture对象
        return new FutureTask<T>(callable);
    }
    

    FutureTask类结构图如下:

    FutureTask是一个异步计算任务,包装了我们外部提交的任务。

    • 实现了Runnable接口
    • 实现了Future接口,该接口封装了任务结果的获取、任务是否结束等接口。

    RunnableFuture类重要属性

    1.任务运行状态state

    // 存储当前任务运行状态
    private volatile int state;
    // 当前任务尚未执行
    private static final int NEW          = 0;
    // 当前任务正在结束,尚未完全结束,一种临界状态
    private static final int COMPLETING   = 1;
    // 当前任务正常结束
    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;
    

    可能的状态转换有如下几种:

    • NEW -> COMPLETING -> NORMAL
    • NEW -> COMPLETING -> EXCEPTIONAL
    • NEW -> CANCELLED
    • NEW -> INTERRUPTING -> INTERRUPTED

    2.真正要执行的任务callble

    // 存放真正提交的原始任务
    private Callable<V> callable;
    

    3.存放执行结果outcome

    返回的结果或从get()中抛出的异常
    private Object outcome;
    

    4.当前正在运行任务的线程runner

    //当前任务被线程执行期间,保存当前任务的线程对象引用
    private volatile Thread runner;
    

    5.调用get获取任务结果的等待线程集合waiters

    //因为会有很多线程去get当前任务的结果,所以这里使用了一种stack数据结构来保存
    private volatile WaitNode waiters;
    
    static final class WaitNode {
            volatile Thread thread;
            volatile WaitNode next;
            WaitNode() { thread = Thread.currentThread(); }
        }
    

    数据结构如下图:

    RunnableFuture类构造方法

    public FutureTask(Callable<V> callable) {
            if (callable == null)
                throw new NullPointerException();
            // 设置要执行的任务
            this.callable = callable;
            // 初始化时任务状态为NEW
            this.state = NEW;       
    }
    

    任务执行run()原理

    submit()方法最终调用线程池的execute()方法,而execute()方法会创建出"工人"Worker对象,调用runWorker()方法,它主要是执行外部提交的任务,也就是这里的FutureTask对象的run()方法, 我们重点看下run()方法。

    FutureTask#run()开始执行任务。

    它主要的功能是完成包装的callable的call方法执行,并将执行结果保存到outcome中,同时捕获了call方法执行出现的异常,并保存异常信息,而不是直接抛出。

    public void run() {
        // 状态机不为NEW表示执行完成或任务被取消了,直接返回
        // 状态机为NEW,同时将runner设置为当前线程,保证同一时刻只有一个线程执行run方法,如果设置失败也直接返回
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            // 取出原始的任务检测不为空 且 再次检查状态为NEW(双重校验)
            if (c != null && state == NEW) {
                // 任务运行的结果
                V result;
                // 任务是否运行是否正常, true:正常, false-异常
                boolean ran;
                try {
                    // 任务执行,将结果返回给result
                    result = c.call();
                    // 设置任务运行正常
                    ran = true;
                } catch (Throwable ex) {
                    // 任务运行报错的情况
                    // 设置结果为空
                    result = null;
                    // 设置任务运行异常标记
                    ran = false;
                    // 任务执行抛出异常时,保存异常信息,而不直接抛出
                    setException(ex);
                }
                // 执行成功则保存结果
                if (ran)
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            // 执行完成后设置runner为null
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            // 获取任务状态
            int s = state;
            // 如果被置为了中断状态则进行中断的处理
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }
    

    FutureTask#set()方法处理正常执行的运行结果

    setException()方法主要完成做下面的工作。

    • 将执行结果保存到outcom变量中
    • FutureTask的状态从NEW修改为NORMAL
    • 唤醒阻塞在waiters队列中请求get的所有线程
    protected void set(V v) {
        // 将状态由NEW更新为COMPLETING
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            // 保存任务的结果
            outcome = v;
           // 更新状态的最终状态-NORMAL
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            // 通用的完成操作,主要作用就是唤醒阻塞在waiters队列中请求get的线程
            finishCompletion();
        }
    }
    

    FutureTask#setException()方法处理执行异常的结果

    setException()方法主要完成做下面的工作。

    • 将异常信息保存到outcom变量中
    • FutureTask的状态从NEW修改为EXCEPTIONAL
    • 唤醒阻塞在waiters队列中请求get的所有线程
    // FutureTask#setException
    protected void setException(Throwable t) {
        // 将状态由NEW更新为COMPLETING
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            // 将异常信息保存到输出结果中
            outcome = t;
            // 更新状态机处理异常的最终状态-EXCEPTIONAL
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            // 通用的完成操作,主要作用就是唤醒阻塞在waiters队列中请求get的线程
            finishCompletion();
        }
    }
    

    这里的finishCompletion()唤醒我们在后面讲解,上面的整个逻辑可以用一张图表示:

    任务结果获取get()原理

    其他线程可以调用get()方法或者超时阻塞方法get(long timeout, TimeUnit unit)获取任务运行的结果。

    FutureTask#get()方法是获取任务执行结果的入口方法。

    // 阻塞获取任务结果
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        // 任务还没有执行完成,通过awaitDone方法进行阻塞等待
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        // 返回结果
        return report(s);
    }
    
    // 超时阻塞获取任务结果
    public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        // 判空处理
        if (unit == null)
            throw new NullPointerException();
        int s = state;
        // 任务还没有执行完成,通过awaitDone方法进行阻塞等待
        if (s <= COMPLETING &&
            // 如果awaitDone返回的结果还是小于等于COMPLETING,表示运行中,那么直接抛出超时异常
            (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
            throw new TimeoutException();
        // 返回结果
        return report(s);
    }
    

    FutureTask#awaitDone()方法阻塞等待任务执行结束

    该方法主要完成下面的工作:

    • 判断任务是否运行结束,结束的话直接返回运行状态
    • 如果任务没有结果,将请求线程阻塞
    • 请求线程阻塞时,会创建一个waiter节点,然后加入到阻塞等待的栈中
    // 线程阻塞等待方法, timed等于 true表示阻塞等待有时间限制nanos, false表示没有,一直阻塞
    private int awaitDone(boolean timed, long nanos) throws InterruptedException {
        // 计算阻塞超时时间点
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        // 表示q是否添加到waiters栈中,默认false
        boolean queued = false;
        // 自旋操作
        for (;;) {
            // 如果阻塞线程被中断则将当前线程从阻塞队列中移除
            if (Thread.interrupted()) {
                // 从waiters栈中移除WaitNode,
                removeWaiter(q);
                // 返回中断移除
                throw new InterruptedException();
            }
    
            // 获取任务的状态
            int s = state;
            // 如果任务的状态大于COMPLETING,表示线程运行结束了,直接返回
            if (s > COMPLETING) { 
                // 任务已经完成时直接返回结果
                if (q != null)
                    q.thread = null;
                // 返回状态
                return s;
            }
            // 如果任务状态是COMPLETING     
            else if (s == COMPLETING) 
                // 如果任务执行完成,但还差最后一步最终完成,则让出CPU给任务执行线程继续执行
                Thread.yield();
            // 如果任务状态小于COMPLETING,说明任务还在运行中
            // 如果q为空的情况    
            else if (q == null)
                // 新进来的线程添加等待节点
                q = new WaitNode();
            // 如果任务还在运行中并且当前线程节点还不在waiters栈中,那么就加入   
            else if (!queued)
                // 上一步节点创建完,还没将其添加到waiters栈中,因此在下一个循环就会执行此处进行入栈操作,并将当前线程的等待节点置于栈顶
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            // 如果任务还在运行中并且timed为true,表示有超时限制
            else if (timed) {
                // 如果设置了阻塞超时时间,则进行检查是否达到阻塞超时时间,达到了则删除当前线程的等待节点并退出循环返回,否则继续阻塞
                nanos = deadline - System.nanoTime();
                // 如果nanos小于等于0
                if (nanos <= 0L) {
                    // 从waiters栈中移除
                    removeWaiter(q);
                    //返回状态
                    return state;
                }
                // 超时阻塞当前线程,超过时间,就会恢复
                LockSupport.parkNanos(this, nanos);
            }
            // 如果任务还在运行中并且timed为false,没有有超时限制
            else
                // 一直阻塞当前线程
                LockSupport.park(this);
        }
    }
    

    FutureTask#report方法解析返回任务结果

    // 获取任务结果方法:正常执行则直接返回结果,否则抛出异常
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        // 如果状态是正常情况
        if (s == NORMAL)
            // 直接返回
            return (V)x;
        // 如果状态是取消了,抛出异常
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }
    

    FutureTask#finishCompletion()方法用来唤醒前面等待的线程

    上一步awaitDone方法会阻塞调用的线程,那么任务运行结束总要唤醒他们去拿结果吧,这个工作就在finishCompletion()方法中。

    private void finishCompletion() {
        // 遍历waiters栈中的每个元素;
        for (WaitNode q; (q = waiters) != null;) {
            // cas设置waiters中q节点数据为null,成功的话,进入到if中
            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
        callable = null;        // to reduce footprint
    }
    

    任务取消cancel()原理

    可以调用FutureTask#cancel方法取消任务执行,但是要注意下面几点:

    • 任务取消时会先检查是否允许取消,当任务已经完成或者正在完成(正常执行并继续处理结果 或 执行异常处理异常结果)时不允许取消。
    • cancel方法有个boolean入参,若为false,则只唤醒所有等待的线程,不中断正在执行的任务线程。若为true则直接中断任务执行线程,同时修改状态为INTERRUPTED。
    // 取消任务,参数mayInterruptIfRunning为true,会中断运行中的线程,false不会
    public boolean cancel(boolean mayInterruptIfRunning) {
        // 如果FutureTask的状态不是NEW或者CAS设置失败时,直接返回false
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {   
            // 如果参数mayInterruptIfRunning为true,中断
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    //cas修改状态为INTERRUPTED
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            // 唤醒其他等待的线程
            finishCompletion();
        }
        return true;
    }
    

    cancel方法实际上完成以下两种状态转换之一:

    • NEW -> CANCELLED (对应于mayInterruptIfRunning=false)
    • NEW -> INTERRUPTING -> INTERRUPTED (对应于mayInterruptIfRunning=true)

    总结

    本文讲解了线程池submit()提交任务的原理实现,通过源码很多平时项目中遇到的坑都找到了答案。比如说之前项目中用线程池submit()方法提交任务处理,发现任务的异常都不见了,这下明白了,原来是通过setException()保存下来了,只有通过get方法获取到,只有看过源码,才会豁然开朗。

    以上就是Java线程池submit阻塞获取结果的实现原理详解的详细内容,更多关于Java线程池submit阻塞获取结果的资料请关注自由互联其它相关文章!

    上一篇:SpringBoot日志配置操作全面介绍
    下一篇:没有了
    网友评论