Java 利用Future異步獲取多線程任務結果

Future接口是Java標准API的一部分，在java.util.concurrent包中。Future接口是Java線程Future模式的實現，可以來進行異步計算。

有了Future就可以進行三段式的編程了，1.啟動多線程任務2.處理其他事3.收集多線程任務結果。從而實現了非阻塞的任務調用。在途中遇到一個問題，那就是雖然能異步獲取結果，但是Future的結果需要通過isdone來判斷是否有結果，或者使用get()函數來阻塞式獲取執行結果。這樣就不能實時跟蹤其他線程的結果狀態了，所以直接使用get還是要慎用，最好配合isdone來使用。

這里有一種更好的方式來實現對任意一個線程運行完成后的結果都能及時獲取的辦法：使用CompletionService，它內部添加了阻塞隊列，從而獲取future中的值，然后根據返回值做對應的處理。一般future使用和CompletionService使用的兩個測試案例如下：

package com.yzuzhang.thread;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 多線程執行，異步獲取結果
 * http://www.cnblogs.com/clarechen/p/4604189.html
 */
public class AsyncThread {
    
    private static List<Future<String>> futureList = new ArrayList<Future<String>>();
    
    public static void main(String[] args) {
        AsyncThread t = new AsyncThread();
        
        t.generate(3);
        t.doOtherThings();
        t.getResult(futureList);
    }

    /**
     * 生成指定數量的線程，都放入future數組
     * 
     * @param threadNum
     * @param fList
     */
    public void generate(int threadNum) {
        ExecutorService service = Executors.newFixedThreadPool(threadNum);
        for (int i = 0; i < threadNum; i++) {
            Callable<String> job = getJob(i);
            Future<String> f = service.submit(job);
            futureList.add(f);
        }
        //關閉線程池,不影響線程的執行
        service.shutdown();
    }

    /**
     * other things
     */
    public void doOtherThings() {
        try {
            for (int i = 0; i < 3; i++) {
                System.out.println("Do thing No:" + i);
                Thread.sleep(new Random().nextInt(5000));
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 從future中獲取線程結果，打印結果
     * 
     * @param fList
     */
    public void getResult(List<Future<String>> fList) {
        ExecutorService service = Executors.newSingleThreadExecutor();
        service.execute(getCollectJob(fList));
        service.shutdown();
    }

    /**
     * 生成指定序號的線程對象
     * 
     * @param i
     * @return
     */
    public Callable<String> getJob(final int i) {
        final int time = new Random().nextInt(10);
        return new Callable<String>() {
            @Override
            public String call() throws Exception {
                Thread.sleep(1000 * time);
                return "thread-" + i;
            }
        };
    }

    /**
     * 生成結果收集線程對象
     * 
     * @param fList
     * @return
     */
    public Runnable getCollectJob(final List<Future<String>> fList) {
        return new Runnable() {
            @Override
            public void run() {
                for (Future<String> future : fList) {
                    try {
                        while (true) {
                            if (future.isDone() && !future.isCancelled()) {
                                System.out.println("Future:" + future + ", Result:" + future.get());
                                break;
                            } else {
                                System.out.println("休眠100ms...");
                                Thread.sleep(100);
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        };
    }

}

運行結果打印和future放入列表時的順序一致，為0，1，2：

Do thing No:0
Do thing No:1
Do thing No:2
Future:java.util.concurrent.FutureTask@591c5342, Result:thread-0
Future:java.util.concurrent.FutureTask@529e3fc2, Result:thread-1
Future:java.util.concurrent.FutureTask@136c03ee, Result:thread-2

 

二、下面是先執行完的線程先處理的方案，使用completionService收集callable結果：

package com.yzuzhang.thread;

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.yzuzhang.utils.SleepUtils;

/**
 * Future就可以進行三段式的編程了，從而實現了非阻塞的任務調用
 * 1.啟動多線程任務 ->處理其他事 ->收集多線程任務結果
 */
public class CompletionCallable {
    //存放處理結果的集合
    private static ConcurrentHashMap<Integer, Integer> map = new ConcurrentHashMap<Integer, Integer>();
    
    public static void main(String[] args) {
        try {
            completionServiceCount();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 使用completionService收集callable結果
     * @throws ExecutionException 
     * @throws InterruptedException 
     */
    public static void completionServiceCount() throws InterruptedException, ExecutionException {
        ExecutorService threadPool = Executors.newCachedThreadPool();
        CompletionService<Integer> cs = new ExecutorCompletionService<Integer>(threadPool);
        
        int threadNum = 5;
        for (int i = 0; i < threadNum; i++) {
            cs.submit(getTask(i));
        }
        int sum = 0, temp = 0;
        for(int i=0;i<threadNum;i++){
            //先執行完的線程,先返回結果
            temp = cs.take().get();
            sum += map.get(temp);
            System.out.print(temp + "\t");
        }
        
        SleepUtils.sleep(1000);
        System.out.println("Result->" + sum);
        threadPool.shutdown();
    }
    
    public static Callable<Integer> getTask(final int no) {
        final Random rand = new Random();
        Callable<Integer> task = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int time = rand.nextInt(100)*100;
                map.put(Integer.valueOf(no), time);
                System.out.println("Thread-"+ no +" sleep :"+ time);
                Thread.sleep(time);
                return no;
            }
        };
        return task;
    }
}

運行結果為最先結束的線程結果先被處理：

Thread-0 sleep :8800
Thread-4 sleep :6800
Thread-1 sleep :4900
Thread-2 sleep :400
Thread-3 sleep :6000
2    1    3    4    0    Result->26900