Executor框架學習筆記

關於ThreadPoolExecutor的源碼解讀，請參考我的最新博客《ThreadPoolExecutor源碼解讀》。

 

Java中的線程即是工作單元也是執行機制，從JDK 5后，工作單元與執行機制被分離。工作單元包括Runnable和Callable，執行機制由JDK 5中增加的java.util.concurrent包中Executor框架提供。

HotSpot VM的線程模型中將java的線程映射為本地操作系統的線程，java線程的啟動意味着一個本地操作系統線程的創建，而java線程的終止也就意味着對應的系統線程的回收。

 

Executor框架主要包含三個部分：

任務：包括Runnable和Callable，其中Runnable表示一個可以異步執行的任務，而Callable表示一個會產生結果的任務

任務的執行：包括Executor框架的核心接口Executor以及其子接口ExecutorService。在Executor框架中有兩個關鍵類ThreadPoolExecutor和ScheduledThreadPoolExecutor實現了ExecutorService接口。

異步計算的結果：包括接口Future和其實現類FutureTask。

 

下面是對Executor框架中的一些關鍵接口與類的簡介

 

Executor接口(java.util.concurrent.Executor)

它是Executor的基礎與核心，其定義如下：

public interface Executor {
    void execute(Runnable command);
}

 

它包含了一個方法execute，參數為一個Runnable接口引用。

Executor接口將任務的提交與執行分離開來。

 

ThreadPoolExecutor類(java.util.concurrent.ThreadPoolExecutor)

它是線程池的核心實現類，用來執行被提交的任務。

它通常由工廠類Executors來創建，Executors可以創建SingleThreadExecutor，FixedThreadPool以及CachedThreadPool等不同的ThreadPoolExecutor。

 

SingleThreadExecutor使用單線程執行任務，Executors提供的API有如下兩個

public static ExecutorService newSingleThreadExecutor();
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory);

SingleThreadExecutor保證了任務執行的順序，不會存在多線程活動。

 

 FixedThreadPool是使用固定線程數的線程池,Executors提供的API有如下兩個

public static ExecutorService newFixedThreadPool(int nThreads);
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory);

FixedThreadPool滿足了資源管理的需求，可以限制當前線程數量。適用於負載較重的服務器環境。

 

CachedThreadPool是無界線程池，Executors提供的API有如下兩個

public static ExecutorService newCachedThreadPool();
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory);

CachedThreadPool適用於執行很多短期異步任務的小程序，適用於負載較輕的服務器。

 

ScheduledThreadPoolExecutor類(java.util.concurrent.ScheduledThreadPoolExecutor)

它是ThreadPoolExecutor的子類且實現了ScheduledExecutorService接口，它可以在給定的延遲時間后執行命令，或者定期執行命令，它比Timer更強大更靈活。

 Executors可以創建的ScheduledThreadPoolExecutor的類型有ScheduledThreadPoolExecutor和SingleThreadScheduledExecutor等

 

ScheduledThreadPoolExecutor具有固定線程個數，適用於需要多個后台線程執行周期任務，並且為了滿足資源管理需求而限制后台線程數量的場景,Executors中提供的API有如下兩個：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory);

 

SingleThreadScheduledExecutor具有單個線程，Executors提供的創建API有如下兩個：

public static ScheduledExecutorService newSingleThreadScheduledExecutor();
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory);

它適用於單個后台線程執行周期任務，並且保證順序一致執行的場景。

 

上述的ThreadPoolExecutor和ScheduledThreadPoolExecutor都可以用於執行Runnable與Callable接口的實現類

 

 

Future接口(Java.concurrent.Future)

Future代表着提交的任務的計算狀態與結果，可以對其進行取消，查詢是否取消，查詢是否完成，查詢結果等操作。

首先來看一下Future接口的定義：

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;

}

 

FutureTask類間接實現了Future接口，它用來表示異步計算的結果。當ThreadPoolExecutor或者ScheduledThreadPoolExecutor執行Runnable接口或者Callable接口的實現類時，它們會返回一個Future接口引用（實現類為FutureTask）。

 

 Runnable接口或者Callable接口的實現類在被上述兩者執行的區別是，前者沒有返回結果，而后者可以返回結果。Runnable可以通過Executors提供的API(Executors#callable)包裝為Callable。

 

 

 

 

本文參考資料：《Java並發編程的藝術》以及其他網上文檔 

 

-----------------------------------------------------------------------------------------------------------------

下面是一些簡單的Demo程序

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class ExecutorTest {
    public static void main(String[] args) {

        Runnable hello = () -> {
            for (int i = 0; i < 100; i++) {
                System.out.println(i + " hello");
            }
        };
        Runnable bye = () -> {
            for (int i = 0; i < 100; i++) {
                System.out.println(i + " bye");
            }
        };

        Executor executor = Executors.newCachedThreadPool();

        executor.execute(hello);
        executor.execute(bye);

    }
}

上面程序使用了兩個Runnable任務hello和bye來打印相應語句，程序將會交錯打印hello和bye。如果將executor改為SingleThreadExecutor，將會先打印100個"hello"，再打印100個"bye"。

 

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

public class ExecutorTest {
    public static void main(String[] args) {
        Random random = new Random();
        List<Integer> numbers = new ArrayList<>();
        for (int i = 0; i < 100000; i++) {
            numbers.add(random.nextInt(100000));
        }
        int result = calculate(numbers, 3);
        System.out.println(result);
    }

    public static int calculate(List<Integer> numbers,int digit) {
        List<Callable<Integer>> tasks = new ArrayList<>();
        for (Integer x : numbers) {
            tasks.add(() -> {
                int count=0;
                int y=x;
                do {
                    if (y % 10 == digit) {
                        count++;
                    }
                    y /= 10;
                } while (y > 0);
                return count;
            });
        }
        ExecutorService service = Executors.newFixedThreadPool(10);
        int answer=0;
        try {
            List<Future<Integer>> results = service.invokeAll(tasks);
            for (Future<Integer> result : results) {
                try {
                    answer+=result.get();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return answer;
    }
}

上面的程序隨機生成了100000個隨機數，然后統計這些數字中每個數字10進制中具有多少個數位3的數量和。使用具有10個線程的線程池進行計算，最終通過Future引用對象的結果來統計答案。