在工程專案中可能會有這麼一個場景，客戶端處理層需要從服務端（CDN/圖片伺服器）獲取n張圖片（參考微博一個人最多有9張圖片），那麼問題來了，如何在一定的時間範圍內儘可能多的獲取到圖片。當然，最為簡單粗暴的方法就是通過序列的方式來獲取，但是如果第一個請求hang住並超時，那麼其他圖片都無法獲取，這顯然不是一個好的設計方式。這個問題設計到兩點，第一點是"儘可能多"，相比於序列，解決方案就是多執行緒並行；第二點是"一定時間範圍內"，也就是超時時間，因此這個問題就可以抽象為多執行緒在超時範圍內獲取資料的方法，下面探討幾張方法。

一、thread.join(long millis) (不滿足需求)

       在JavaDoc中join(time)功能為線上程退出之前將會最多等待time毫秒。事實上，如果超時時間過了，那麼主執行緒將會繼續停止阻塞（可見join方法是阻塞的）並繼續執行，這麼做有一個弊端可能會造成所謂的"執行緒洩露"。此外，我們通常會呼叫interrupt來中斷執行緒，當然這麼做也是不提倡的。

TimeoutDemo.java：

public class TimeoutDemo {

    public static void main(String[] args) {

        try {

            int threadCount = 3;

            int defaultTimes = 10;

            Thread[] ts = new Thread[threadCount];

            for (int i=0; i < threadCount; i++) {
                ts[i] = new Thread(new TimeoutRunnable(defaultTimes));
            }

            for (int i=0; i < threadCount; i++) {
                ts[i].start();
            }

            for (int i=0; i < threadCount; i++) {
                ts[i].join(1000);
                System.out.println("i = " + i + " has joined...");
            }

            for (int i=0; i < threadCount; i++) {
                ts[i].interrupt();
            }

        } catch (Exception e) {
            System.out.println("thread interrupted when waiting join.");
        } finally {
            System.out.println("all threads finished...");
        }
    }
}

class TimeoutRunnable implements Runnable{

    private int times;

    public TimeoutRunnable(int times) {
        this.times = times;
    }

    @Override
    public void run() {
        for (int i = 0; i < times; i++) {
            System.out.println("thread id: " + Thread.currentThread().getId() + " runs for " + (i+1) + "th time.");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println("thread id: " + Thread.currentThread().getId() + " is interrupted while running and stop right now.");

                return;
            }
        }

        System.out.println("thread id: " + Thread.currentThread().getId() + " finished.");
    }
}
 

執行結果：

thread id: 11 runs for 1th time.
thread id: 13 runs for 1th time.
thread id: 12 runs for 1th time.
i = 0 has joined...
thread id: 12 runs for 2th time.
thread id: 11 runs for 2th time.
thread id: 13 runs for 2th time.
i = 1 has joined...
thread id: 11 runs for 3th time.
thread id: 12 runs for 3th time.
thread id: 13 runs for 3th time.
i = 2 has joined...
thread id: 13 runs for 4th time.
thread id: 11 runs for 4th time.
thread id: 12 runs for 4th time.
thread id: 11 is interrupted while running and stop right now.
thread id: 13 is interrupted while running and stop right now.
all threads finished...
thread id: 12 is interrupted while running and stop right now.
 

可見，join方法是阻塞的，依次等待執行緒超時，就本例來講，等待了3個1000ms，不滿足在一定的timeout內完成。

二、基於ExecutorService的shutdown() 和 awaitTermination(long timeout, TimeUnit unit)方法：

        awaitTermination是一個阻塞方法，在ExecutorService執行完shutdown之後進行操作，awaitTermination將會阻塞，直到所有任務完成了執行，或者超時產生，或者執行緒中斷髮生為止。如果所有任務完成則返回true；如果在任務完成之前發生了超時則返回false。程式碼示例，TimeoutDemo1.java：

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TimeoutDemo1 {

    public static void main(String[] args) {

        try {

            int threadCount = 10;

            int defaultTimes = 10;

            ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

            for (int i=0; i < threadCount * 2; i++) {
                executorService.submit(new TimeoutRunnable1(defaultTimes));
            }

            executorService.shutdown();

            if(!executorService.awaitTermination(7, TimeUnit.SECONDS)) {
                System.out.println("executors did not terminate in the specified time.");
                List<Runnable> droppedTasks = executorService.shutdownNow();
                System.out.println("executors was abruptly shut down. " + droppedTasks.size() + " tasks will not be executed.");
            }
        } catch (Exception e) {
            System.out.println("exception: " + e.getMessage());
        } finally {
            System.out.println("scheduled executor service closed normally.");
        }
    }
}

class TimeoutRunnable1 implements Runnable{

    private int times;

    public TimeoutRunnable1(int times) {
        this.times = times;
    }

    @Override
    public void run() {
        for (int i = 0; i < times; i++) {
            System.out.println("thread id: " + Thread.currentThread().getId() + " runs for " + (i+1) + "th time.");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println("thread id: " + Thread.currentThread().getId() + " is interrupted while running and stop right now.");

                return;
            }
        }
        System.out.println("thread id: " + Thread.currentThread().getId() + " finished.");
    }
}

關於shutdown和shutdonwNow的區別參考前一篇文章：如何優雅的關閉執行緒池。

三、使用CountDownLatch中的await(long timeout, TimeUnit unit)

    如果在timeout時間範圍內，count變成了0，那麼await返回true；如果超過了超時時間，count不為0，那麼返回false。

看程式碼TimeoutDemo2.java：

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TimeoutDemo2 {

    public static void main(String[] args) {

        try {

            int threadCount = 10;

            int defaultTimes = 10;

            CountDownLatch countDownLatch = new CountDownLatch(threadCount * 2);

            ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

            for (int i=0; i < threadCount * 2; i++) {
                executorService.submit(new TimeoutRunnable2(defaultTimes, countDownLatch));
            }

            if(!countDownLatch.await(19, TimeUnit.SECONDS)) {
                executorService.shutdown();
                System.out.println("executors did not terminate in the specified time.");
                List<Runnable> droppedTasks = executorService.shutdownNow();
                System.out.println("executors was abruptly shut down. " + droppedTasks.size() + " tasks will not be executed.");
            }
        } catch (Exception e) {
            System.out.println("exception: " + e.getMessage());
        } finally {
            System.out.println("scheduled executor service closed normally.");
        }
    }
}

class TimeoutRunnable2 implements Runnable{

    private int times;

    private CountDownLatch countDownLatch;

    public TimeoutRunnable2(int times,  CountDownLatch countDownLatch) {
        this.times = times;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < times; i++) {
                System.out.println("thread id: " + Thread.currentThread().getId() + " runs for " + (i + 1) + "th time.");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    System.out.println("thread id: " + Thread.currentThread().getId() + " is interrupted while running and stop right now.");

                    return;
                }
            }

            System.out.println("thread id: " + Thread.currentThread().getId() + " finished.");
        } catch (Exception e) {
            System.out.println(e.getMessage());
        } finally {
            countDownLatch.countDown();
        }
    }
}

四、Future.get(long timeout, TimeUnit unit)（不完全滿足需求，不最優）

      future.get在一定的超時時間範圍內獲取執行緒執行的返回值，但是get方法是阻塞的，會在一定的超時時間之內阻塞，直到任務完成返回，或者這時間超時丟擲TimeoutException，程式碼如下：

TimeoutDemo3.java:

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class TimeoutDemo3 {

    public static void main(String[] args) {

        try {
            int threadCount = 10;

            int defaultTimes = 10;

            ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

            List<Future> futureList = new ArrayList<Future>();

            for (int i=0; i < threadCount; i++) {
                Future future = executorService.submit(new TimeoutRunnable3(defaultTimes));
                futureList.add(future);
            }

            for (Future future: futureList) {
                String str = (String)future.get(15, TimeUnit.SECONDS);

                System.out.println(str);
            }
        } catch (InterruptedException e) {
            System.out.println("future is interrupted...");
        } catch (ExecutionException e) {
            System.out.println("future execution exception...");
        } catch (TimeoutException e) {
            System.out.println("future timeout exception...");
        } finally {
            System.out.println("scheduled executor service closed normally.");
        }
    }
}

class TimeoutRunnable3 implements Callable<String>{

    private int times;

    public TimeoutRunnable3(int times) {
        this.times = times;
    }

    @Override
    public String call() {
        try {
            for (int i = 0; i < times; i++) {
                System.out.println("thread id: " + Thread.currentThread().getId() + " runs for " + (i + 1) + "th time.");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    System.out.println("thread id: " + Thread.currentThread().getId() + " is interrupted while running and stop right now.");
                }
            }

            //System.out.println("thread id: " + Thread.currentThread().getId() + " finished.");
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }

        return "thread id: " + Thread.currentThread().getId() + " finished";
    }
}

Author：憶之獨秀

Email：[email protected]