Java：多線程，線程池，使用CompletionService通過Future來處理Callable的返回結果

1. 背景

在Java5的多線程中，可以使用Callable接口來實現具有返回值的線程。使用線程池的submit方法提交Callable任務，利用submit方法返回的Future存根，調用此存根的get方法來獲取整個線程池中所有任務的運行結果。

方法一：如果是自己寫代碼，應該是自己維護一個Collection保存submit方法返回的Future存根，然后在主線程中遍歷這個Collection並調用Future存根的get()方法取到線程的返回值。

方法二：使用CompletionService類，它整合了Executor和BlockingQueue的功能。你可以將Callable任務提交給它去執行，然后使用類似於隊列中的take方法獲取線程的返回值。

2. 實現代碼

package com.clzhang.sample.thread;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

public class ThreadPoolTest4 {
    // 具有返回值的測試線程
    class MyThread implements Callable<String> {
        private String name;
        public MyThread(String name) {
            this.name = name;
        }

        @Override
        public String call() {
            int sleepTime = new Random().nextInt(1000);
            try {
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 返回給調用者的值
            String str = name + " sleep time:" + sleepTime;
            System.out.println(name + " finished...");

            return str;
        }
    }

    private final int POOL_SIZE = 5;
    private final int TOTAL_TASK = 20;

    // 方法一，自己寫集合來實現獲取線程池中任務的返回結果
    public void testByQueue() throws Exception {
        // 創建線程池
        ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
        BlockingQueue<Future<String>> queue = new LinkedBlockingQueue<Future<String>>();

        // 向里面扔任務
        for (int i = 0; i < TOTAL_TASK; i++) {
            Future<String> future = pool.submit(new MyThread("Thread" + i));
            queue.add(future);
        }

        // 檢查線程池任務執行結果
        for (int i = 0; i < TOTAL_TASK; i++) {
            System.out.println("method1:" + queue.take().get());
        }

        // 關閉線程池
        pool.shutdown();
    }

    // 方法二，通過CompletionService來實現獲取線程池中任務的返回結果
    public void testByCompetion() throws Exception {
        // 創建線程池
        ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
        CompletionService<String> cService = new ExecutorCompletionService<String>(pool);

        // 向里面扔任務
        for (int i = 0; i < TOTAL_TASK; i++) {
            cService.submit(new MyThread("Thread" + i));
        }

        // 檢查線程池任務執行結果
        for (int i = 0; i < TOTAL_TASK; i++) {
            Future<String> future = cService.take();
            System.out.println("method2:" + future.get());
        }

        // 關閉線程池
        pool.shutdown();
    }

    public static void main(String[] args) throws Exception {
        ThreadPoolTest4 t = new ThreadPoolTest4();
        t.testByQueue();
        t.testByCompetion();
    }
}

部分輸出：

...

Thread4 finished...
method1:Thread4 sleep time:833
method1:Thread5 sleep time:158
Thread6 finished...
method1:Thread6 sleep time:826
method1:Thread7 sleep time:185
Thread9 finished...
Thread8 finished...
method1:Thread8 sleep time:929
method1:Thread9 sleep time:575

...

Thread11 finished...
method2:Thread11 sleep time:952
Thread18 finished...
method2:Thread18 sleep time:793
Thread19 finished...
method2:Thread19 sleep time:763
Thread16 finished...
method2:Thread16 sleep time:990

...

3. 總結

使用方法一，自己創建一個集合來保存Future存根並循環調用其返回結果的時候，主線程並不能保證首先獲得的是最先完成任務的線程返回值。它只是按加入線程池的順序返回。因為take方法是阻塞方法，后面的任務完成了，前面的任務卻沒有完成，主程序就那樣等待在那兒，只到前面的完成了，它才知道原來后面的也完成了。

使用方法二，使用CompletionService來維護處理線程不的返回結果時，主線程總是能夠拿到最先完成的任務的返回值，而不管它們加入線程池的順序。