1. Thread 和 Runnable

1.1 Thread

我們先看一下 Thread 最簡(jiǎn)單的使用姿勢(shì)：

public class MyThread extends Thread {
    public MyThread(String name) {
        super(name);
    }
    @Override
    public void run() {
        String name = Thread.currentThread().getName();
        System.out.println(name + "已經(jīng)運(yùn)行");

    }
    public static void main(String[] args) {
        new MyThread("線程一").start();
    }
}

線程包含 4 個(gè)狀態(tài)：創(chuàng)建 -> 就緒 -> 運(yùn)行 -> 結(jié)束。

當(dāng)執(zhí)行 start() 后，線程進(jìn)入就緒狀態(tài)，當(dāng)對(duì)應(yīng)的線程搶占到 cpu 調(diào)度資源之后，進(jìn)入運(yùn)行狀態(tài)，此時(shí)調(diào)用的是 run 方法，執(zhí)行完畢之后就結(jié)束了。

1.2 Runnable

我們看一下 Runnable 最簡(jiǎn)單的使用姿勢(shì)：

public class MyTask implements Runnable {
    @Override
    public void run() {
        String name = Thread.currentThread().getName();
        System.out.println(name + "已經(jīng)運(yùn)行");
    }
    public static void main(String[] args) {
        new Thread(new MyTask(),"線程二").start();
    }
}

這里 MyTask 就是一個(gè) Runnable，實(shí)現(xiàn)了 run() 方法，作為 Thread() 的入?yún)ⅰ?/p>

基本所有同學(xué)都知道這樣使用，但是你們知道原理么？

1.3 Thread 和 Runnable 的關(guān)系

我們看一下 Runnable 的接口定義：

public interface Runnable {
    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see     java.lang.Thread#run()
     */
    public abstract void run();
}

英文翻譯大致如下：當(dāng)一個(gè)對(duì)象繼承并實(shí)現(xiàn)了 run() 方法，當(dāng)線程 start() 后，會(huì)在該線程中單獨(dú)執(zhí)行該對(duì)象的 run() 方法。

這段翻譯，基本就告訴了 Runnable 和 Thread 的關(guān)系：

1. MyTask 繼承 Runnable，并實(shí)現(xiàn)了 run() 方法；
2. Thread 初始化，將 MyTask 作為自己的成員變量；
3. Thread 執(zhí)行 run() 方法，線程處于“就緒”狀態(tài)；
4. 等待 CPU 調(diào)度，執(zhí)行 Thread 的 run() 方法，但是 run() 的內(nèi)部實(shí)現(xiàn)，其實(shí)是執(zhí)行的 MyTask.run() 方法，線程處于“運(yùn)行”狀態(tài)。

這里面的第2、4步，需要對(duì)照著源碼看看。

在 Thread 初始化時(shí)，MyTask 作為入?yún)?target，最后賦值給 Thread.target：

當(dāng)執(zhí)行 Thread.run() 時(shí)，其實(shí)是執(zhí)行的 target.run()，即 MyTask.run()，這個(gè)是典型的策略模式：

2. Callable 、Future 和 FutureTask

先看一下它們的整體關(guān)系圖譜：

我剛開始看到這幅圖，感覺 Java 真是麻煩，已經(jīng)有了 Thread 和 Runnable 這兩種創(chuàng)建線程的方式，為啥又搞這 3 個(gè)東西呢？

其實(shí)對(duì)于 Thread 和 Runable，其 run() 都是無返回值的，并且無法拋出異常，所以當(dāng)你需要返回多線程的數(shù)據(jù)，就需要借助 Callable 和 Future。

2.1 Callable

Callable 是一個(gè)接口，里面有個(gè) V call() 方法，這個(gè) V 就是我們的返回值類型：

public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

我們一般會(huì)用匿名類的方式使用 Callable，call() 中是具體的業(yè)務(wù)邏輯：

Callable<String> callable = new Callable<String>() {
    @Override
    public String call() throws Exception {
        // 執(zhí)行業(yè)務(wù)邏輯 ...
        return "this is Callable is running";
    }
};

這里拋出一個(gè)問題，這個(gè) callable.call() 和 Thread.run() 是什么關(guān)系呢？

2.2 FutureTask

通過關(guān)系圖譜，F(xiàn)utureTask 繼承了 RunnableFuture，RunnableFuture 繼承了 Runnable 和 Future：

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

所以，F(xiàn)utureTask 也是個(gè) Runnable ?。?！

這里就有點(diǎn)意思了，既然 FutureTask 是個(gè) Runnable，肯定就需要實(shí)現(xiàn) FutureTask.run() 方法，那么 FutureTask 也可以作為 Thread 的初始化入?yún)ⅲ褂米藙?shì)如下：

new Thread(FutureTask對(duì)象).start();

所以當(dāng)執(zhí)行 Thread.run() 時(shí)，其實(shí)是執(zhí)行的 FutureTask.run()，這個(gè)是我們破解的第一層。

下面我們?cè)倨平?FutureTask.run() 和 Callable.call() 的關(guān)系。

2.3 Callable 和 FutureTask 的關(guān)系

FutureTask 初始化時(shí)，Callable 必須作為 FutureTask 的初始化入?yún)ⅲ?/p>

當(dāng)執(zhí)行 FutureTask.run() 時(shí)，其實(shí)執(zhí)行的是 Callable.call()：

所以，這里又是一個(gè)典型的策略模式 ?。。?/strong>

現(xiàn)在我們應(yīng)該可以很清楚知道 Thread 、Runnable、FutureTask 和 Callable 的關(guān)系：

• Thread.run() 執(zhí)行的是 Runnable.run()；
• FutureTask 繼承了 Runnable，并實(shí)現(xiàn)了 FutureTask.run()；
• FutureTask.run() 執(zhí)行的是 Callable.run()；
• 依次傳遞，最后 Thread.run()，其實(shí)是執(zhí)行的 Callable.run()。

所以整個(gè)設(shè)計(jì)方法，其實(shí)就是 2 個(gè)策略模式，Thread 和 Runnable 是一個(gè)策略模式，F(xiàn)utureTask 和 Callable 又是一個(gè)策略模式，最后通過 Runnable 和 FutureTask 的繼承關(guān)系，將這 2 個(gè)策略模式組合在一起。

嗯嗯。。。我們是不是把 Future 給忘了~~

2.4 Future

為什么要有 Future 呢？我再問一個(gè)問題，大家可能就知道了。

我們通過 FutureTask，借助 Thread 執(zhí)行線程后，結(jié)果數(shù)據(jù)我們?cè)趺传@取到呢？這里就需要借助到 Future。

我們看一下 Future 接口：

public interface Future<V> {
    // 取消任務(wù)，如果任務(wù)正在運(yùn)行的，mayInterruptIfRunning為true時(shí)，表明這個(gè)任務(wù)會(huì)被打斷的，并返回true；
    // 為false時(shí)，會(huì)等待這個(gè)任務(wù)執(zhí)行完，返回true；若任務(wù)還沒執(zhí)行，取消任務(wù)后返回true，如任務(wù)執(zhí)行完，返回false
    boolean cancel(boolean mayInterruptIfRunning);
    // 判斷任務(wù)是否被取消了,正常執(zhí)行完不算被取消
    boolean isCancelled();
    // 判斷任務(wù)是否已經(jīng)執(zhí)行完成，任務(wù)取消或發(fā)生異常也算是完成，返回true
    boolean isDone();
    // 獲取任務(wù)返回結(jié)果，如果任務(wù)沒有執(zhí)行完成則等待完成將結(jié)果返回，如果獲取的過程中發(fā)生異常就拋出異常，
    // 比如中斷就會(huì)拋出InterruptedException異常等異常
    V get() throws InterruptedException, ExecutionException;
    // 在規(guī)定的時(shí)間如果沒有返回結(jié)果就會(huì)拋出TimeoutException異常
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

對(duì)于 FutureTask，Callable 就是他的任務(wù)，而 FutureTask 內(nèi)部維護(hù)了一個(gè)任務(wù)狀態(tài)，所有的狀態(tài)都是圍繞這個(gè)任務(wù)來進(jìn)行的，隨著任務(wù)的進(jìn)行，狀態(tài)也在不斷的更新。

FutureTask 繼承了 Future，實(shí)現(xiàn)對(duì)任務(wù)的取消、數(shù)據(jù)獲取、任務(wù)狀態(tài)判斷等功能。

比如我們經(jīng)常會(huì)調(diào)用 get() 方法獲取數(shù)據(jù)，如果任務(wù)沒有執(zhí)行完成，會(huì)將當(dāng)前線程放入阻塞隊(duì)列等待，當(dāng)任務(wù)執(zhí)行完后，會(huì)喚醒阻塞隊(duì)列中的線程。

3. 具體實(shí)例

private static List<String> processByMultiThread(Integer batchSize) throws ExecutionException, InterruptedException {
    List<String> output = new ArrayList<>();

    // 獲取分批數(shù)據(jù)
    List<List<Integer>> batchProcessData = getProcessData(batchSize);

    // 啟動(dòng)線程
    List<FutureTask<List<String>>> futureTaskList = new ArrayList<>();
    for (List<Integer> processData : batchProcessData) {
        Callable<List<String>> callable = () -> processOneThread(processData);
        FutureTask<List<String>> futureTask = new FutureTask<>(callable);
        new Thread(futureTask).start();  // 啟動(dòng)線程
        futureTaskList.add(futureTask);
    }

    // 獲取線程返回的數(shù)據(jù)
    for (FutureTask futureTask : futureTaskList) {
        List<String> processData = (List<String>) futureTask.get();
        output.addAll(processData);
    }
    return output;
}

這個(gè)示例很簡(jiǎn)單：

1. 先將數(shù)據(jù)按照 batchSize 分成 N 批；
2. 啟動(dòng) N 個(gè)線程，去執(zhí)行任務(wù)；
3. 通過 futureTask.get() 獲取每個(gè)線程數(shù)據(jù)，并匯總輸出。

這個(gè)示例其實(shí)不太適合線上的場(chǎng)景，因?yàn)槊看握{(diào)用都會(huì)初始化線程，如果調(diào)用過多，內(nèi)存可能會(huì)被撐爆，需要借助線程池。

完整的示例代碼，以及線程池使用姿勢(shì)，詳見github：https://github.com/lml200701158/java-study/blob/master/src/main/java/com/java/parallel/share/MultiThreadProcess.java

··················END················