並發編程（十二）—— Java 線程池 實現原理與源碼深度解析 之 submit 方法 （二）

在上一篇《並發編程（十一）—— Java 線程池 實現原理與源碼深度解析（一）》中提到了線程池ThreadPoolExecutor的原理以及它的execute方法。這篇文章是接着上一篇文章寫的，如果你沒有閱讀上一篇文章，建議你去讀讀。本文解析ThreadPoolExecutor#submit。

　　對於一個任務的執行有時我們不需要它返回結果，但是有我們需要它的返回執行結果。對於線程來講，如果不需要它返回結果則實現Runnable，而如果需要執行結果的話則可以實現Callable。在線程池同樣execute提供一個不需要返回結果的任務執行，而對於需要結果返回的則可調用其submit方法。

AbstractExecutorService

我們把上一篇文章的代碼貼過來

public abstract class AbstractExecutorService implements ExecutorService {

    // RunnableFuture 是用於獲取執行結果的，我們常用它的子類 FutureTask
    // 下面兩個 newTaskFor 方法用於將我們的任務包裝成 FutureTask 提交到線程池中執行
    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);
    }

    // 提交任務
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        // 1. 將任務包裝成 FutureTask
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        // 2. 交給執行器執行，execute 方法由具體的子類來實現
        // 前面也說了，FutureTask 間接實現了Runnable 接口。
        execute(ftask);
        return ftask;
    }

    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        // 1. 將任務包裝成 FutureTask
        RunnableFuture<T> ftask = newTaskFor(task, result);
        // 2. 交給執行器執行
        execute(ftask);
        return ftask;
    }

    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        // 1. 將任務包裝成 FutureTask
        RunnableFuture<T> ftask = newTaskFor(task);
        // 2. 交給執行器執行
        execute(ftask);
        return ftask;
    }
}

 

盡管submit方法能提供線程執行的返回值，但只有實現了Callable才會有返回值，而實現Runnable的線程則是沒有返回值的，也就是說在上面的3個方法中，submit(Callable<T> task)能獲取到它的返回值，submit(Runnable task, T result)能通過傳入的載體result間接獲得線程的返回值或者准確來說交給線程處理一下，而最后一個方法submit(Runnable task)則是沒有返回值的，就算獲取它的返回值也是null。

使用示例

submit(Callable<T> task)

/**
 * @author: ChenHao
 * @Date: Created in 14:54 2019/1/11
 */
public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Callable<String> callable = new Callable<String>() {
            public String call() throws Exception {
                System.out.println("This is ThreadPoolExetor#submit(Callable<T> task) method.");
                return "result";
            }
        };

        ExecutorService executor = Executors.newSingleThreadExecutor();
        Future<String> future = executor.submit(callable);
        executor.shutdown();
        System.out.println(future.get());
    }
}

運行結果：

submit(Runnable task, T result)

/**
 * @author: ChenHao
 * @Date: Created in 14:54 2019/1/11
 */
public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ExecutorService executor = Executors.newSingleThreadExecutor();
        Data data = new Data();
        Future<Data> future = executor.submit(new Task(data), data);
        executor.shutdown();
        System.out.println(future.get().getName());
    }
}
class Data {
    String name;
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
}

class Task implements Runnable {
    Data data;
    public Task(Data data) {
        this.data = data;
    }
    @Override
    public void run() {
        System.out.println("This is ThreadPoolExetor#submit(Runnable task, T result) method.");
        data.setName("陳浩");
    }
}

 運行結果：

submit(Runnable task)

/**
 * @author: ChenHao
 * @Date: Created in 14:54 2019/1/11
 */
public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                System.out.println("This is ThreadPoolExetor#submit(Runnable runnable) method.");
            }
        };

        ExecutorService executor = Executors.newSingleThreadExecutor();
        Future future = executor.submit(runnable);
        executor.shutdown();
        System.out.println(future.get());
    }
}

運行結果：

 

從上面的源碼可以看到，這三者方法幾乎是一樣的，關鍵就在於：

RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);

是如何將一個任務作為參數傳遞給了newTaskFor，然后調用execute方法，最后進而返回ftask的呢？

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

 

源碼分析

這里我建議大家去看看我之前的一篇文章《Java 多線程（五）—— 線程池基礎 之 FutureTask源碼解析》

submit(Callable<T> task)

我們看上面的源碼中知道實際上是調用了如下代碼

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

 

 我們看看 FutureTask 的結構

public class FutureTask<V> implements RunnableFuture<V> { 
    private volatile int state; 
    private static final int NEW = 0; //初始狀態 
    private static final int COMPLETING = 1; //結果計算完成或響應中斷到賦值給返回值之間的狀態。 
    private static final int NORMAL = 2; //任務正常完成，結果被set 
    private static final int EXCEPTIONAL = 3; //任務拋出異常 
    private static final int CANCELLED = 4; //任務已被取消 
    private static final int INTERRUPTING = 5; //線程中斷狀態被設置ture，但線程未響應中斷 
    private static final int INTERRUPTED = 6; //線程已被中斷 

    //將要執行的任務 
    private Callable<V> callable; //用於get()返回的結果，也可能是用於get()方法拋出的異常 
    private Object outcome; // non-volatile, protected by state reads/writes //執行callable的線程，調用FutureTask.run()方法通過CAS設置 
    private volatile Thread runner; //棧結構的等待隊列，該節點是棧中的最頂層節點。 
    private volatile WaitNode waiters; 

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

 

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

 

 

 我們知道 FutureTask 繼承了 Runnable，這里將 Callable<T> callable 的實例封裝成 FutureTask 傳給 execute(ftask);我們再來看看上一篇文章中線程運行的代碼

// 此方法由 worker 線程啟動后調用，這里用一個 while 循環來不斷地從等待隊列中獲取任務並執行
// 前面說了，worker 在初始化的時候，可以指定 firstTask，那么第一個任務也就可以不需要從隊列中獲取
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 該線程的第一個任務(如果有的話)
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 循環調用 getTask 獲取任務
        while (task != null || (task = getTask()) != null) {
            w.lock();          
            // 如果線程池狀態大於等於 STOP，那么意味着該線程也要中斷
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 到這里終於可以執行任務了，這里是最重要的，task是什么？是Worker 中的firstTask屬性
                    
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                // 一個任務執行完了，這個線程還可以復用，接着去隊列中拉取任務執行
                // 置空 task，准備 getTask 獲取下一個任務
                task = null;
                // 累加完成的任務數
                w.completedTasks++;
                // 釋放掉 worker 的獨占鎖
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 如果到這里，需要執行線程關閉：
        // 說明 getTask 返回 null，也就是超過corePoolSize的線程過了超時時間還沒有獲取到任務，也就是說，這個 worker 的使命結束了，執行關閉
        processWorkerExit(w, completedAbruptly);
    }
}

 

 由上面第6行代碼 task 就是execute(ftask)傳入的任務，第26行 task.run(); 實際上就是 new FutureTask<T>(callable).run(),我們看看FutureTask中的run()方法

public void run() {
    //保證callable任務只被運行一次
    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 { 
                //執行任務，上面的例子我們可以看出，call()里面可能是一個耗時的操作，不過這里是同步的
                result = c.call();
                //上面的call()是同步的，只有上面的result有了結果才會繼續執行
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
                //執行完了，設置result
                set(result);
        }
    }
    finally {
        runner = null;
        int s = state;
        //判斷該任務是否正在響應中斷，如果中斷沒有完成，則等待中斷操作完成
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

 

在 FutureTask的構造方法中 this.callable = callable; ，因此我們可以知道上面run()方法中第6行 c 就是 代碼示例中的 new Callable<String>()，c.call()就是調用 代碼示例中new Callable 的call方法，並且這里可以取到返回結果，第22行處設置FutureTask 中 outcome 的值，這樣線程就可以取到返回值了。

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}

 

取值我就不分析了，我之前的文章里面有詳細分析。

submit(Runnable task, T result)

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

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

 

我們來看看FutureTask的另外一個構造方法

public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

 

public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}

static final class RunnableAdapter<T> implements Callable<T> {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

 

上面將 runnable, result 封裝成了 RunnableAdapter 作為FutureTask的callable屬性，這和上面的submit(Callable<T> task) 是不同的，submit(Callable<T> task)是直接將 Callable<T> task作為FutureTask的callable屬性。我們看看FutureTask中的run()方法中第6行 c 就是FutureTask 構造方法中的new RunnableAdapter<T>(task, result) ，c.call()就是調用 RunnableAdapter<T>(task, result) 的call方法，call()中的task.run()就是上面代碼示例中new Task(data) 中的 run()，run()方法中業務大代碼改變了data對象的屬性，callable(Runnable task, T result)中也是傳的相同的對象data, 所以，result = c.call(); 就是把更改后的data返回，並且將data設置為設置FutureTask 中 outcome 的值，后面的邏輯就是一樣的了。

這里可以看成將同一個data傳入線程進行處理，同時這個data也傳入FutureTask中，並且在RunnableAdapter通過屬性進行保存data,等線程將data處理完了，由於是同一個對象，RunnableAdapter中的result也就是data指向的是同一個對象，然后把此result返回到FutureTask保存在屬性outcome中，就可以通過FutureTask.get()取到運行結果了。

如果new FutureTask<T>(runnable, null),則result = c.call(); 返回的值也是null,最后從線程池中get的值也是null。