多執行緒併發修改一個數據結構，很容易破壞這個資料結構，如散列表。鎖能夠保護共享資料結構，但選擇執行緒安全的實現更好更容易，如阻塞佇列就是執行緒安全的集合。

執行緒安全的集合

Vector和HashTable類提供了執行緒安全的動態陣列和散列表，而ArrayList和HashMap卻不是執行緒安全的。

java.util.concurrent包提供了對映表、有序集、佇列的高效實現，如：

• ConcurrentLinkedQueue：多執行緒安全訪問，無邊界，非阻塞，佇列；

• ConcurrentHashMap：多執行緒安全訪問，雜湊對映表，初始容量預設16，調整因子預設0.75。

併發的雜湊對映表ConcurrentHashMap

對於執行緒安全的集合，返回的是弱一致性的迭代器：

• 迭代器不一定能反映出構造後的所有修改；

• 迭代器不會將同一個值返回兩次；

• 迭代器不會丟擲ConcurrentModificationException異常。

通常執行緒安全的集合能夠高效的支援大量的讀者和一定數量的寫者，當寫者執行緒數目大於設定值時，後來的寫者執行緒會被暫時阻塞。而對於大多數執行緒安全的集合，size()方法一般無法在常量時間完成，一般需要遍歷整個集合才能確定大小。

同步包裝器

任何集合類使用同步包裝器都會變成執行緒安全的，會將集合的方法使用鎖加以保護，保證執行緒的安全訪問。使用同步包裝器時要確保沒有任何執行緒通過原始的非同步方法訪問資料結構，也可以說確保不存在任何指向原始物件的引用，可以採用下面構造一個集合並立即傳遞給包裝器的方法定義。

List<E> synchArrayList = Collections.synchronizedList(new
 
 ArrayList<E>());
Map<K, V> synchHashMap = Collections.synchronizedMap(new HashMap<K, V>());

當然最好使用java.util.concurrent包中定義的集合，同步包裝器並沒有太多安全和效能上的優勢。

Callable與Future

Callable與Runnable類似，都可以封裝一個非同步執行的任務，但是Callable有返回值。Callabele<T>介面是一個引數化的型別，只有一個方法call()，型別引數就是返回值的型別。Future用來儲存非同步計算的結果，用get()方法獲取結果。get()方法的呼叫會被阻塞，直到計算完成。有超時引數的get()方法超時時會丟擲TimeoutException異常。

FutureTask可將Callable轉換成Future和Runnable，實現了兩者的介面。

Callable<Integer> myComputation = new MyComputationCallable();
FutureTask<Integer> task = new FutureTask<Integer>(myComputation);
Thread t = new Thread(task);  // it's a Runnable
t.start();
Integer result = task.get();  // it's a Future

這裡有一個計算指定目錄及其子目錄下與關鍵字匹配的檔案數目的例子，涉及到Callable、FutureTask、Future的使用。

public Integer call() {
    count = 0;
    try {
        File [] files = directory.listFiles();
        List<Future<Integer>> results = new ArrayList<>();

        for (File file : files) {
            if (file.isDirectory()) {
                MatchCounter counter = new MatchCounter(file, keyword);
                FutureTask<Integer> task = new FutureTask<>(counter);
                results.add(task);
                Thread t = new Thread(task);
                t.start();
            } else {
                if (search(file)) {
                    count++;
                }
            }
        }

        for (Future<Integer> result : results) {
            try {
                count += result.get();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    } catch (InterruptedException e) {
        ;
    }
    return count;
}

執行緒池

構建一個新的執行緒是有代價的，涉及到與作業系統的互動。對於程式中需要建立大量生命期很短的執行緒，應該使用執行緒池。執行緒池中的執行緒執行完畢並不會馬上死亡，而是在池中準備為下一個請求提供服務。當然使用執行緒池還可以限制併發執行緒的數目。

需要呼叫執行器Executors的靜態工廠方法來構建執行緒池，下面的方法返回的是ExecutorService介面的ThreadPoolExecutor類的物件。

• Executors.newCachedThreadPool：執行緒空閒60秒後終止，若有空閒執行緒立即執行任務，若無則建立新執行緒。

• Executors.newFixedThreadPool：池中執行緒數由引數指定，固定大小，剩餘任務放置在佇列。

使用submit()方法，將Runnable物件或Callable物件提交給執行緒池ExecutorService，任務何時執行由執行緒池決定。呼叫submit()方法，會返回一個Future物件，用來查詢任務狀態或結果。當用完執行緒池時，要記得呼叫shutdown()關閉，會在所有任務執行完後徹底關閉。類似的呼叫shutdownNow，可取消尚未開始的任務並試圖終端正在執行的執行緒。

執行緒池的使用步驟大致如下：

1. 呼叫Executors類的靜態方法newCachedThreadPool()或newFixedThreadPool()；

2. 呼叫submit()提交Runnable或Callable物件；

3. 如果提交Callable物件，就要儲存好返回的Future物件；

4. 執行緒池用完時，呼叫shutdown()。

對於之前提到的計算檔案匹配數的例子，需要產生大量生命期很多的執行緒，可以使用一個執行緒池來執行任務，完整程式碼在這裡。

public Integer call() {
    count = 0;
    try {
        File [] files = directory.listFiles();
        List<Future<Integer>> results = new ArrayList<>();
        for (File file : files) {
            if (file.isDirectory()) {
                MatchCounter counter = new MatchCounter(file, keyword, pool);
                Future<Integer> result = pool.submit(counter);
                results.add(result);
            } else {
                if (search(file)) {
                    count++;
                }
            }
        }
        for (Future<Integer> result : results) {
            try {
                count += result.get();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    } catch (InterruptedException e) {
        ;
    }
    return count;
}

Fork-Join框架

對於多執行緒程式，有些應用使用了大量執行緒，但其中大多數都是空閒的。還有些應用需要完成計算密集型任務，Fork-Join框架專門用來支援這類任務。使用Fork-Join框架解決思路大致是分治的思想，採用遞迴計算再合併結果。只需繼承RecursiveTask<T>類，並覆蓋compute()方法。invokeAll()方法接收很多工並阻塞，直到這些任務完成，join()方法將生成結果。

對於問題，統計陣列中滿足某特性的元素個數，使用Fork-Join框架是很合適的。

import java.util.concurrent.*;

public class ForkJoinTest {
    public static void main(String [] args) {
        final int SIZE = 10000000;
        double [] numbers = new double[SIZE];
        for (int i = 0; i < SIZE; i++) {
            numbers[i] = Math.random();
        }
        Counter counter = new Counter(numbers, 0, numbers.length, new Filter() {
            public boolean accept(double x) {
                return x > 0.5;
            }
        });
        ForkJoinPool pool = new ForkJoinPool();
        pool.invoke(counter);
        System.out.println(counter.join());
    }
}

interface Filter {
    boolean accept(double t);
}

class Counter extends RecursiveTask<Integer> {
    private final int THRESHOLD = 1000;
    private double [] values;
    private int from;
    private int to;
    private Filter filter;

    public Counter(double [] values, int from, int to, Filter filter) {
        this.values = values;
        this.from = from;
        this.to = to;
        this.filter = filter;
    }

    public Integer compute() {
        if (to - from < THRESHOLD) {
            int count = 0;
            for (int i = from; i < to; i++) {
                if (filter.accept(values[i])) {
                    count++;
                }
            }
            return count;
        } else {
            int mid = (from + to) / 2;
            Counter first = new Counter(values, from, mid, filter);
            Counter second = new Counter(values, mid, to, filter);
            invokeAll(first, second);
            return first.join() + second.join();
        }
    }
}

另外，Fork-Join框架使用工作密取來平衡可用執行緒的工作負載，比手工多執行緒強多了。

轉：https://segmentfault.com/a/1190000004918145