Future任務機制： Fork/Join 框架

Future任務機制和FutureTask

化繁為簡，分而治之，遞歸的分解和合並，直到任務小到可以接受的程度。

Fork/Join 框架是Java7提供的一個用於並行執行任務的框架。

是一個把大任務分割成若干個小任務，最終匯總每個小任務結果后得到大任務結果的框架。

Fork 就是把大任務切分為若干子任務並行的執行

Join 就是合並這些子任務的執行結果，最后得到這個大任務的結果。

public interface Callable<V> {
    
    // 只有一個獲得返回結果的方法,實現這個方法即可
    
    V call() throws Exception;
}

Future類就是對於具體的Runnable或者Callable任務的執行結果進行取消、查詢是否完成、獲取結果。

必要時，通過get方法獲取執行結果，該方法會阻塞直到任務返回結果。

package路徑：java.util.concurrent。也是一個接口。

public interface Future<V> {

    // 1. 如果取消成功則返回true, 取消失敗則返回false
    // 2. mayInterruptIfRunning 表示是否允許取消正在執行卻沒有完畢的任務,如果設置true, 則取消。
    // 3. 如果任務完成, mayInterruptIfRunning為true,還是false, 都返回false。
    // 4. 如果取消已經完成的任務返回false
    // 5. 如果任務正在執行,若mayInterruptIfRunning設置為true, 則返回true。若mayInterruptIfRunning設置為false, 則返回false。
    // 6. 如果任務還沒有執行,則無論mayInterruptIfRunning設置為true還是false, 肯定返回true
    boolean cancel(boolean mayInterruptIfRunning);
    
    // 表示任務是否被取消成功,如果在任務正常完成前被取消,則返回ture
    boolean isCanceled();
    
    // 表示任務是否已經完成, 若任務完成, 則返回true。
    boolean isDone();
    
    // 用來獲取執行結果, 這個方法會產生阻塞, 會一直等到任務執行完畢才返回
    V get() throws InterruptedException, ExecutionException;
    
    // 用來獲取執行結果, 如果在指定時間內, 還沒獲取到結果, 就直接返回null
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

Future 提供了三種功能：

（1）判斷任務是否完成

（2）能夠中斷任務

（3）能夠獲取任務執行結果

 

FutureTask是Future接口的一個唯一實現類。

public class FutureTask implements RunnableFuture<V> {

}

// RunnableFuture 接口的實現
public interface RunnableFuture <V>extends Runnable, Future<V> {

    void run();
}

FutureTask實現了RunnableFuture，所以它既可以作為Runnable被線程執行，又可以作為Future得到Callback的返回值。

兩個構造函數：

public FutureTask(Callable<V> callable) {
        // 創建一個FutureTask, 一旦運行就執行給定的Callback。
    }
    
    public FutureTask(Runnable runnable, V result) {
        // 創建一個FutureTask, 一旦運行就執行給定的runnable, 並安排成功完成時,get返回給定的結果。
    }

接下來的Fork/Join就是基於Future實現的：

class demo {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        FoundTask task1 = new FoundTask("Thread Found Name");
        FutureTask<String> f1 = new FutureTask<String>(task1);

        new Thread(f1).start();

        System.out.printf(f1.get());

        FutureTask<Integer> f2 = new FutureTask<Integer>(new FoundRun(), 2);

        new Thread(f2).start();
        System.out.printf("result-" + f2.get());
    }
}

class FoundTask implements Callable<String> {

    private String mName = null;

    FoundTask(String name) {
        mName = name;
    }

    @Override
    public String call() throws Exception {

        Thread.sleep(1000);

        System.out.printf(mName + "finish the task");

        return "result-1";
    }
}

class FoundRun implements Runnable {

    @Override
    public void run() {

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.printf("FoundRun finish");
    }
}

運行結果：

Thread Found Name finish the task
result-1
FoundRun finish
result-2

按照預想，按順序，按步驟的執行。 

使用Fork/Join框架，首先要考慮到的是如何分割任務:

e.g.

class demo {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        ForkJoinPool forkJoinPool = new ForkJoinPool();

        CountTask task = new CountTask(1, 5);

        Future<Integer> result = forkJoinPool.submit(task);

        System.out.printf("1-5最終相加的結果:" + result.get());

        CountTask task2 = new CountTask(1, 100);

        Future<Integer> result2 = forkJoinPool.submit(task2);

        System.out.printf("1-100最終相加的結果:" + result2.get());
        
        // end
    }
}

class CountTask extends RecursiveTask<Integer> {

    private static final long serialVersionUID = 3336021421713606929L;

    private static int splitSize = 2;
    private int mStart, mEnd;

    public CountTask(int start, int end) {
        mStart = start;
        mEnd = end;
    }


    @Override
    protected Integer compute() {
        int sum = 0;
        // 如果任務已經不需要再拆分了就開始計算。
        boolean canCompute = (mEnd - mStart) <= splitSize;

        if (canCompute) {

            for (int i = mStart; i <= mEnd; i++) {

                sum = sum + i;
            }
        } else {

            // 拆分兩個字任務。
            int middle = (mStart - mEnd) / 2;
            CountTask firstTask = new CountTask(mStart, middle);
            CountTask secondTask = new CountTask(middle + 1, mEnd);

            firstTask.fork(); // 開始執行
            secondTask.fork();

            int firstResult = firstTask.join(); // 獲得第一個子任務結果,得不到結果,此線程不會往下面執行。
            int secondResult = secondTask.join();

            // 合並兩個子任務的結果。
            sum = firstResult + secondResult;
        }

        return sum;
    }
}

運行結果：

1-5最終相加的結果:15
1-100最終相加的結果::5050

 

Fork/Join模式優缺點及應用場景。

注意：分拆的對象過多時，小心一下子把內存撐滿了，等待線程的CPU資源釋放了，但是線程對象等待時，不會被垃圾機制回收。

場景：

對於樹形結構類型的數據的處理和遍歷非常適合。

e.g. 我們要對一個靜態資源服務器的圖片文件目錄進行遍歷和分析的時候，我們需要遞歸的統計每個目錄下的文件數量，最后匯總，非常適合用分叉／結合框架來處理。

class demo {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        Integer count = new ForkJoinPool().invoke(new CountingTask(Paths.get("D://fish")));
        
        System.out.printf("D:盤fish下面總文件數量:" + count);
        
        // end
    }
}

// 處理單個目錄的任務
class CountingTask extends RecursiveTask<Integer> {

    private Path mDir;
    public CountingTask (Path dir) {
        mDir = dir;
    }


    @Override
    protected Integer compute() {
        
        int count = 0;
        List<CountingTask> subTasks = new ArrayList<>();
        
        // 讀取目錄dir的子路徑。
        try {
            
            DirectoryStream<Path> ds = Files.newDirectoryStream(dir);
            
            for (Path subPath : ds) {
                
                if (Files.isDirectory(subPath, LinkOption.NOFOLLOW_LINKS)) {
                    
                    // 對每個子目錄都新建一個子任務
                    subTasks.add(new CountingTask(subPath));
                } else {
                    
                    // 遇到文件,則計數器增加1
                    count ++;
                }
            }
            
            if (!subTasks.isEmpty()) {
                // 在當前的ForkJoinPool上調度所有的子任務
                for (CountingTask subTask : invokeAll(subTasks)) {
                    
                    count += subTask.join();
                }
            }
            
        } catch (IOException ex) {
            return 0;
        }
    }
}

運行結果：運算速度還是非常快的，但是一旦文件多了，也是非常耗資源的，電腦就會出現卡頓的情況

D:盤fish下面總文件數量:7647