ThreadPoolExecutor中的submit()方法详细讲解

墨蓝 2022-06-14 04:09 216阅读 0赞

## ThreadPoolExecutor中的submit()方法详细讲解 ##

在使用线程池的时候，发现除了`execute()`方法可以执行任务外，还发现有一个方法`submit()`可以执行任务。

`submit()`有3个参数不一的方法，这些方法都是在`ExecutorService`接口中声明的，在`AbstractExecutorService`中实现，而`ThreadPoolExecutor`继承`AbstractExecutorService`。

<T> Future<T> submit(Callable<T> callable);
    
    <T> Future<T> submit(Runnable var1, T result);
    
    Future<?> submit(Runnable runnable);

我们可以看到`submit()`的参数既可以是`Runnable`，又可以是`Callable`。对于`Runnable`我们是比较熟的，它是线程`Thread`所执行的任务，里面有一个`run()`方法，是任务的具体执行操作。那么`Callable`呢？我们一起看下他们的代码吧。

public interface Runnable {
        void run();
    }
    
    
    public interface Callable<V> {
        V call() throws Exception;
    }

`Runnable`这里就不介绍了，`Callable`接口定义了一个`call()`方法，返回一个`Callable`指定的泛型类，并且`call()`调用的时候会抛出异常。通过比较`Runnable`和`Callable`还看不什么端倪，那么我们就看看内部实现吧。

*  **`submmit()`参数解析**

这里重点分析`submit()`带参数`Runnable`和`Callable`的方法

public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
    
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

我们发现2者的实现没有任何的差异，唯一就是`submit()`参数不同。

参数传入`newTaskFor()`方法，那么可以肯定就是在这个方法里做了什么操作。

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

`newTaskFor()`的目的就是创建一个`FutureTask`对象，那我们追踪到`FutureTask`的构造方法(`FutureTask`非常关键，后面会分析)。

public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       
    }
    
    
    public FutureTask(Callable<V> callable) {
        if (callable == null)throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       
    }

到了这里我们知道，其实`Runnable`会在这里转化成`Callable`。我们来看下`Executors.callable()`具体实现。

public static <T> Callable<T> callable(Runnable task, T result) {
        if (task == null)
            throw new NullPointerException();
        return new RunnableAdapter<T>(task, result);
    }
    
    
    private static final class RunnableAdapter<T> implements Callable<T> { 
        private final Runnable task;
        private final T result;
        RunnableAdapter(Runnable task, T result) {
            this.task = task;
            this.result = result;
        }
        public T call() {
            task.run();
            return result;
        }
    }

`Executors.callable()`创建了一个`RunnableAdapter`对象，`RunnableAdapter`实现了`Callable`接口，在`call()`方法中调用了传入的`Runnable`的`run()`，并且将传入的`result`参数返回。

也就是说我们调用`submit()`传入的`Runnbale`最终会转化成`Callable`，并且返回一个`result`值(如果我们传入这个参数则返回这个参数，不传入则返回null)。

到这里我们讲清楚了`submit()`的参数的区别和内部实现，`submit()`方法有一个返回值`Future`，下面我们来分析一下返回值`Future`。

*  **`submit()`的返回值`Future`**

上面分析`submit()`源码可知，`submit()`返回的是一个`RunnableFuture`类对象，真正是通过`newTaskFor()`方法返回一个`new FutureTask()`对象。所以`submit()`返回的真正的对象是`FutureTask`对象。

那么`FutureTask`是什么，我们来看下它的类继承关系。

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

通过继承关系我们可以明确的知道其实`FutureTask`就是一个`Runnable`。并且有自己`run()`实现。我们来看下`FutureTask`的`run()`是如何实现的。

public void run() {
            if (state != NEW ||
                !U.compareAndSwapObject(this, RUNNER, 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)
                        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);
            }
        }

我们在`new FutureTask()`对象的时候，在`FutureTask`构造方法中会对`state`状态赋值为`NEW`，并且传入一个`callable`对象。通过`FutureTask`的`run()`我们可以知道，其实就通过`state`状态判断，调用`callable的call()`。(如果传入的参数是`Runnable`，`Runnable`在`RunnableAdapter`类中转化时，在`call()`中，其实调用的就是`Runnable`的`run()`方法)。

所以在`submit()`方法中，调用了一个`execute(task)`的方法，实际执行的是`FutureTask`的`run()`，而`FutureTask`的`run()`调用的是`Callable`的`call()`方法。

说了这么多，`submit()`最后执行的还是传入的`Runnable`的`run()`或`Callable`的`call()`方法。好像没有`FutureTask`什么事啊。

其实不是，`submit()`返回`FutureTask`对象，通过这个`FutureTask`对象调用`get()`可以返回`submit()`方法传入的一个泛型类参数`result`对象，如果是`Callable`直接通过`call()`返回。这个返回值的可以用来校验任务执行是否成功。

*  **FutureTask的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;
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }

最后是通过`outcome`参数将根据任务的状态将结果返回。那么`outcome`参数在哪里赋值了？`outcome`参数赋值的地方有好2处，一是`FutureTask`的`set()`，二是`FutureTask`的`setException()`。

`set()`是在`FutureTask`的`run()`执行完成后，将传入的`result`参数赋值给传入给`set()`，赋值给`outcome`参数。如果`run()`报异常了会将`Throwable`对象通过`setException()`方法传入，赋值给`outcome`变量

大家可以返回上面的`run()`查看下。

protected void set(V v) {
        if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) {
            outcome = v;
            U.putOrderedInt(this, STATE, NORMAL); // final state
            finishCompletion();
        }
    }
    
    
    protected void setException(Throwable t) {
        if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) {
            outcome = t;
            U.putOrderedInt(this, STATE, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

*  **`submit()`使用案例**

public class Test {
        private static final String SUCCESS = "success";
    
        public static void main(String[] args) {
    
            ExecutorService executorService = Executors.newFixedThreadPool(3);
    
            System.out.println("------------------任务开始执行---------------------");
    
            Future<String> future = executorService.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    Thread.sleep(5000);
                    System.out.println("submit方法执行任务完成" + " thread name: " + Thread.currentThread().getName());
                    return SUCCESS;
                }
            });
    
            try {
                String s = future.get();
                if (SUCCESS.equals(s)) {
                    String name = Thread.currentThread().getName();
                    System.out.println("经过返回值比较，submit方法执行任务成功 thread name: " + name);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
    
            System.out.println("-------------------main thread end---------------------");
        }
    }

打印结果：

------------------任务开始执行--------------------- call()调用开始: 1496899867882
    submit方法执行任务完成: 1496899872897   thread name: pool-1-thread-1
    经过返回值比较，submit方法执行任务成功    thread name: main
    -------------------main thread end---------------------

主线程会一直阻塞，等待线程池中的任务执行完后，在执行后面的语句。