一文帶你了解J.U.C的FutureTask、Fork/Join框架和BlockingQueue

2022 年 6 月 10 日
筆記
JAVA, 並發, 程式設計師之家, FutureTask, J.U.C, BlockingQueue

摘要： J.U.C是Java並發編程中非常重要的工具包，今天，我們就來著重講講J.U.C裡面的FutureTask、Fork/Join框架和BlockingQueue。

本文分享自華為雲社區《【高並發】J.U.C組件擴展》，作者：冰河。

FutureTask

FutureTask是J.U.C(java.util.concurrent)下的，但不是AQS(AbstractQueuedSynchronizer)的子類。其對執行緒結果的處理值得借鑒和在項目中使用。

Thread和Runnable執行完任務無法獲取執行結果。Java1.5開始提供了Callable和Future，通過它們可以在任務執行完畢之後，得到任務執行的結果。

Callable與Runnable介面對比

Callable：泛型介面，提供一個call()方法，支援拋出異常，並且執行後有返回值
Runnable：介面，提供一個run()方法，不支援拋出異常，執行後無返回值

Future介面

對於具體的Callable和Runnable任務，可以進行取消，查詢任務是否被取消，查詢是否完成以及獲取結果等。

Future可以監視目標執行緒調用call()的情況，當調用Future的get()方法時，就可以獲得結果。此時，執行任務的執行緒可能不會直接完成，當前執行緒就開始阻塞，直到call()方法結束返回結果，當前執行緒才會繼續執行。總之，Future可以得到別的執行緒任務方法的返回值。

FutureTask類

實現的介面為RunnableFuture，而RunnableFuture介面繼承了Runnable和Future兩個介面，所以FutureTask類最終也是執行Callable類型的任務。如果FutureTask類的構造方法參數是Runnable的話，會轉換成Callable類型。

類實現了兩個介面：Runnable和Future。所以，它即可以作為Runnable被執行緒執行，又可以作為Future得到Callable的返回值，這樣設計的好處如下：

假設有一個很費時的邏輯，需要計算並且返回這個值，同時，這個值又不是馬上需要，則可以使用Runnable和Future的組合，用另外一個執行緒去計算返回值，而當前執行緒在使用這個返回值之前，可以做其他的操作，等到需要這個返回值時，再通過Future得到。

Future示例程式碼如下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@Slf4j
public class FutureExample {
    static class MyCallable implements Callable<String>{
        @Override
        public String call() throws Exception {
            log.info("do something in callable");
            Thread.sleep(5000);
            return "Done";
        }
    }
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Future<String> future = executorService.submit(new MyCallable());
        log.info("do something in main");
        Thread.sleep(1000);
        String result = future.get();
        log.info("result: {}", result);
        executorService.shutdown();
    }
}

FutureTask示例程式碼如下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
@Slf4j
public class FutureTaskExample {
    public static void main(String[] args) throws Exception{
        FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                log.info("do something in callable");
                Thread.sleep(5000);
                return "Done";
            }
        });
        new Thread(futureTask).start();
        log.info("do something in main");
        Thread.sleep(1000);
        String result = futureTask.get();
        log.info("result: {}", result);
    }
}

Fork/Join框架

位於J.U.C(java.util.concurrent)中，是Java7中提供的用於執行並行任務的框架，其可以將大任務分割成若干個小任務，最終匯總每個小任務的結果後得到最終結果。基本思想和Hadoop的MapReduce思想類似。

主要採用的是工作竊取演算法（某個執行緒從其他隊列里竊取任務來執行），並行分治計算中的一種Work-stealing策略

為什麼需要使用工作竊取演算法呢？

假如我們需要做一個比較大的任務，我們可以把這個任務分割為若干互不依賴的子任務，為了減少執行緒間的競爭，於是把這些子任務分別放到不同的隊列里，並為每個隊列創建一個單獨的執行緒來執行隊列里的任務，執行緒和隊列一一對應，比如A執行緒負責處理A隊列里的任務。但是有的執行緒會先把自己隊列里的任務幹完，而其他執行緒對應的隊列里還有任務等待處理。幹完活的執行緒與其等著，不如去幫其他執行緒幹活，於是它就去其他執行緒的隊列里竊取一個任務來執行。而在這時它們會訪問同一個隊列，所以為了減少竊取任務執行緒和被竊取任務執行緒之間的競爭，通常會使用雙端隊列，被竊取任務執行緒永遠從雙端隊列的頭部拿任務執行，而竊取任務的執行緒永遠從雙端隊列的尾部拿任務執行。

工作竊取演算法的優點：

充分利用執行緒進行並行計算，並減少了執行緒間的競爭

工作竊取演算法的缺點：

在某些情況下還是存在競爭，比如雙端隊列里只有一個任務時。並且該演算法會消耗更多的系統資源，比如創建多個執行緒和多個雙端隊列。

對於Fork/Join框架而言，當一個任務正在等待它使用Join操作創建的子任務結束時，執行這個任務的工作執行緒查找其他未被執行的任務，並開始執行這些未被執行的任務，通過這種方式，執行緒充分利用它們的運行時間來提高應用程式的性能。為了實現這個目標，Fork/Join框架執行的任務有一些局限性。

Fork/Join框架局限性：

（1）任務只能使用Fork和Join操作來進行同步機制，如果使用了其他同步機制，則在同步操作時，工作執行緒就不能執行其他任務了。比如，在Fork/Join框架中，使任務進行了睡眠，那麼，在睡眠期間內，正在執行這個任務的工作執行緒將不會執行其他任務了。
（2）在Fork/Join框架中，所拆分的任務不應該去執行IO操作，比如：讀寫數據文件
（3）任務不能拋出檢查異常，必須通過必要的程式碼來出來這些異常

Fork/Join框架的核心類

Fork/Join框架的核心是兩個類：ForkJoinPool和ForkJoinTask。ForkJoinPool負責實現工作竊取演算法、管理工作執行緒、提供關於任務的狀態以及執行資訊。ForkJoinTask主要提供在任務中執行Fork和Join操作的機制。

示例程式碼如下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
@Slf4j
public class ForkJoinTaskExample extends RecursiveTask<Integer> {
    public static final int threshold = 2;
    private int start;
    private int end;
    public ForkJoinTaskExample(int start, int end) {
        this.start = start;
        this.end = end;
    }
    @Override
    protected Integer compute() {
        int sum = 0;
        //如果任務足夠小就計算任務
        boolean canCompute = (end - start) <= threshold;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 如果任務大於閾值，就分裂成兩個子任務計算
            int middle = (start + end) / 2;
            ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle);
            ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end);

            // 執行子任務
            leftTask.fork();
            rightTask.fork();

            // 等待任務執行結束合併其結果
            int leftResult = leftTask.join();
            int rightResult = rightTask.join();

            // 合併子任務
            sum = leftResult + rightResult;
        }
        return sum;
    }
    public static void main(String[] args) {
        ForkJoinPool forkjoinPool = new ForkJoinPool();

        //生成一個計算任務，計算1+2+3+4
        ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100);

        //執行一個任務
        Future<Integer> result = forkjoinPool.submit(task);

        try {
            log.info("result:{}", result.get());
        } catch (Exception e) {
            log.error("exception", e);
        }
    }
}

BlockingQueue

阻塞隊列，是執行緒安全的。

被阻塞的情況如下：

（1）當隊列滿時，進行入隊列操作
（2）當隊列空時，進行出隊列操作

使用場景如下：

主要在生產者和消費者場景

BlockingQueue的方法

BlockingQueue 具有 4 組不同的方法用於插入、移除以及對隊列中的元素進行檢查。如果請求的操作不能得到立即執行的話，每個方法的表現也不同。這些方法如下：

四組不同的行為方式解釋:

拋出異常

如果試圖的操作無法立即執行，拋一個異常。

特殊值

如果試圖的操作無法立即執行，返回一個特定的值(常常是 true / false)。

阻塞

如果試圖的操作無法立即執行，該方法調用將會發生阻塞，直到能夠執行。

超時

如果試圖的操作無法立即執行，該方法調用將會發生阻塞，直到能夠執行，但等待時間不會超過給定值。返回一個特定值以告知該操作是否成功(典型的是 true / false)。

BlockingQueue的實現類如下：

ArrayBlockingQueue：有界的阻塞隊列（容量有限，必須在初始化的時候指定容量大小，容量大小指定後就不能再變化），內部實現是一個數組，以FIFO的方式存儲數據，最新插入的對象是尾部，最新移除的對象是頭部。
DelayQueue：阻塞的是內部元素，DelayQueue中的元素必須實現一個介面——Delayed（存在於J.U.C下）。Delayed介面繼承了Comparable介面，這是因為Delayed介面中的元素需要進行排序，一般情況下，都是按照Delayed介面中的元素過期時間的優先順序進行排序。應用場景主要有：定時關閉連接、快取對象、超時處理等。內部實現使用PriorityQueue和ReentrantLock。
LinkedBlockingQueue：大小配置是可選的，如果初始化時指定了大小，則是有邊界的；如果初始化時未指定大小，則是無邊界的（其實默認大小是Integer類型的最大值）。內部實現時一個鏈表，以FIFO的方式存儲數據，最新插入的對象是尾部，最新移除的對象是頭部。
PriorityBlockingQueue：帶優先順序的阻塞隊列，無邊界，但是有排序規則，允許插入空對象(也就是null)。所有插入的對象必須實現Comparable介面，隊列優先順序的排序規則就是按照對Comparable介面的實現來定義的。可以從PriorityBlockingQueue中獲得一個迭代器Iterator，但這個迭代器並不保證按照優先順序的順序進行迭代。
SynchronousQueue：隊列內部僅允許容納一個元素，當一個執行緒插入一個元素後，就會被阻塞，除非這個元素被另一個執行緒消費。因此，也稱SynchronousQueue為同步隊列。SynchronousQueue是一個無界非快取的隊列。準確的說，它不存儲元素，放入元素只有等待取走元素之後，才能再次放入元素

點擊關注，第一時間了解華為雲新鮮技術~

Tags: JAVA 並發程式設計師之家 FutureTask J.U.C BlockingQueue