并发专题-Executor源码详解

作者: pq217 | 来源:发表于2022-06-06 20:08 被阅读0次

    Runnable && Thread

    Runnable和Thread都是java.lang包最基本的线程操作类,相当于官方的,而Executor接口及其实现都是Doug Lea写的java.util.concurrent包下,属于民间的,当然因为太牛逼了所以也在jdk中

    先看官方提供的线程操作,其中Runnable是一个函数式接口

    @FunctionalInterface
    public interface Runnable {
        public abstract void run();
    }
    

    可以理解为一个待执行的函数,或者理解为一个任务(通过调用run方法可以实际的执行任务)

    Runable是一个定义的任务,而Thread是它的一个执行者,它提供start方法可以开启一个新线程执行传入的Runable任务,这很像命令模式,用户通过实现Runable制定一个命令,交给Thread这个执行者去具体执行

    Runnable && Thread

    所以一般开启新线程执行方法的方式如下

    Runnable task = () -> {
        // do something
    };
    new Thread(task).start(); // 开启新线程执行
    

    而开启新线程执行方法也只有这一个途径可走,就是必须通过官方的Thread.start方法

    Executor

    虽然开启线程执行任务只能走Thread.start方法,方法只有一个,但我们能做的是可以改变任务运行的方式,比如我们可以决定什么时候执行任务,多少个任务共用某个线程排队工作

    最具代表性的就是线程池,线程池只是修改了任务执行的方式:即所有任务共用固定数量的线程,但最终的运行终归还是通过Thread.start方法实际在线程中执行Runnable方法

    Doug Lea所写的java.util.concurrent.Executor把各种方式的Runnable执行器的一个抽象

    public interface Executor {
        void execute(Runnable command);
    }
    
    Executor

    而且提供了一些常用的执行器供我们使用,比如ThreadPoolExecutor(线程池),ForkJoinPool,当然我们也可以自己定义一个Executor按照自己的方式执行任务,比如netty中实现的SingleThreadEventExecutor是一种单个线程依次处理所有任务的执行器

    ExecutorService

    如果说Executor是一种对按自己套路执行任务的执行器,是一种抽象分类,那么ExecutorService就是其下的一个子分类,它是一种特殊的执行器,从名字直译来看:"执行器服务",从一个执行器升级为执行服务,像不像某公司从卖产品业务升级为产品安装售后整套服务

    所以ExecutorService作为一个特殊的服务类Executor,不光能按照自己的方式执行任务,还推出了一系列附加"服务",那就看看这种特殊的执行器都提供了什么服务

    public interface ExecutorService extends Executor {
    
        void shutdown();
        
        Future<?> submit(Runnable task);
        
        <T> Future<T> submit(Callable<T> task);
        
        ......
    }
    

    只贴了些重要方法,首先继承了Executor肯定是要继承void execute(Runnable) 方法代表它首先是一个任务执行器

    shutdown方法代表这个执行器是有状态的,可以关闭服务的

    然后就是重量级的submit方法,这也是ExecutorService提供的最具特色的服务,如果打开ExecutorService的类,注释第一句就写着:

    /**
     * An {@link Executor} that provides methods to manage termination and
     * methods that can produce a {@link Future} for tracking progress of
     * one or more asynchronous tasks.
    

    翻译过来大致就是ExecutorService是一个特殊的Executor执行器,他可以终止服务并且可以创建一个Future来跟踪任务执行进度

    ExecutorService的submit不光能接受Runnable,还可以接受一种新型任务形式:Callable,即有返回结果和异常的任务

    Callable

    上面我们总结Runnable是一种可执行任务,而这种任务是没有返回结果的,也不能抛出异常,很显然现实中很多任务是需要有返回结果的,比如计算1+1等于几的任务,所以为了扩展任务类型,Doug Lea又定义一种新的任务:Callable,而ExecutorService可以接受并处理这样的任务

    @FunctionalInterface
    public interface Callable<V> {
        V call() throws Exception;
    }
    
    Future

    submit执行的返回值是一个Future,从注释看出他可以跟踪任务的进度,它就好比任务的一个订单,通过订单可以取消任务,查看任务进度等

    • boolean cancel(boolean) 取消
      取消任务,就好比在某宝买了个东西,本质就是提交一个"把东西给我送过来"的任务,而通过订单我们就可以取消这个任务
    • isCancelled() && isDone()
      查看任务状态,是否取消和是否完成
    • V get()
      获取任务执行结果,如果没完成则阻塞,如果是Runnable,返回的就是null
    ExecutorService

    AbstractExecutorService

    ExecutorService制订了一种新型执行器,它的特殊在于可以跟踪任务进度甚至取消任务,那么如何实现呐

    首先execute方法只会单纯的执行任务,按照自己执行器的逻辑调用Thread.start方法,不会有返回值,也不支持跟踪进度或取消

    所以解决方案只有一个:调包任务,具体这样操作:当用户提交任务,不是直接去execute执行,而是把任务包装为一个新任务,新任务执行原任务的同时,还负责获取任务结果,跟踪任务状态等工作,相当于是原任务的一个代理

    以上即是AbstractExecutorService负责的工作,它是ExecutorService关于任务跟踪业务的相关实现(解决方案即是代理任务),继承了AbstractExecutorService的任务执行器既可以实现跟踪任务的功能

    AbstractExecutorService

    来看一下AbstractExecutorService的submit方法

    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        // 生成一个依赖于原任务的新任务
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        // 执行新任务
        execute(ftask);
        // 返回新任务(充当任务跟踪器)
        return ftask;
    }
    

    其中newTaskFor方法负责生成新任务,同时也是原任务的跟踪器(订单)

    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }
    

    所以整个执行过程就是把Runnable或Callable转换为FutureTask的过程,而FutureTask首先是一个新任务(继承Runnable),又是原任务的跟踪器(继承Future),这种任务归类为RunnableFuture

    public interface RunnableFuture<V> extends Runnable, Future<V> {
        void run();
    }
    
    FutureTask

    接下来就看FutureTask是如何实现跟踪替换原任务的,首先它的重点属性如下

    • int state; 存储原任务的执行状态
    • Callable<V> callable; 原任务Runnable也可以适配为返回null的Callable
    • Object outcome; 原任务的返回结果

    再看一下重点方法

    1.初始化,存储原任务,任务状态为NEW

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
    

    2.run(),真实的被执行任务,加入状态判断,执行原任务通过try-catch获取异常,通过outcome保存结果

    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 {
                    // 执行原任务
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    // 有异常设置异常
                    setException(ex);
                }
                if (ran)
                    // 设置outcome存储返回值
                    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);
        }
    }
    

    3.get(),获取结果,如果状态未完成阻塞等待,完成则返回outcome

    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;
        if (s == NORMAL)
            return (V)x; // 返回结果即outcome
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }
    

    4.cancel(),只要把状态设置为CANCEL,run时就会直接return而不会执行原任务

    public boolean cancel(boolean mayInterruptIfRunning) {
        // 状态变为CANCELLED
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        // 如果正在运行调用interrupt阻断正在执行的任务
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            finishCompletion();
        }
        return true;
    }
    
    总结

    AbstractExecutorService只是实现了让任务变得可跟踪,通过调包任务,而具体任务的执行最终依然会调用execute,这个方法AbstractExecutorService并没有实现,因为这也是所有执行器的差异所在,即按自己的方式选择线程执行任务, 比如ThreadPoolExecutor线程池的固定线程数执行所有任务的模式,也就是只需要实现execute方法,而submit则交给父类AbstractExecutorService处理

    相关文章

      网友评论

        本文标题:并发专题-Executor源码详解

        本文链接:https://www.haomeiwen.com/subject/oxdjmrtx.html