本文從代碼審查過程中發(fā)現(xiàn)的一個 ArrayList 相關(guān)的「線程安全」問題出發(fā)，來剖析和理解線程安全。

案例分析

前兩天在代碼 Review 的過程中，看到有小伙伴用了類似以下的寫法：

List<String> resultList = new ArrayList<>();

paramList.parallelStream().forEach(v -> {
    String value = doSomething(v);
    resultList.add(value);
});

印象中 ArrayList 是線程不安全的，而這里會多線程改寫同一個 ArrayList 對象，感覺這樣的寫法會有問題，于是看了下 ArrayList 的實現(xiàn)來確認(rèn)問題，同時復(fù)習(xí)下相關(guān)知識。

先貼個概念：

線程安全 是程式設(shè)計中的術(shù)語，指某個函數(shù)、函數(shù)庫在多線程環(huán)境中被調(diào)用時，能夠正確地處理多個線程之間的共享變量，使程序功能正確完成。 ——維基百科

我們來看下 ArrayList 源碼里與本話題相關(guān)的關(guān)鍵信息：

public class ArrayList<E> extends AbstractList<E>
        implements List<E>, RandomAccess, Cloneable, java.io.Serializable
{
    // ...
    
    /**
     * The array buffer into which the elements of the ArrayList are stored.
     * The capacity of the ArrayList is the length of this array buffer... 
     */
    transient Object[] elementData; // non-private to simplify nested class access

    /**
     * The size of the ArrayList (the number of elements it contains).
     */
    private int size;

    // ...

    /**
     * Appends the specified element to the end of this list...
     */
    public boolean add(E e) {
        ensureCapacityInternal(size + 1);  // Increments modCount!!
        elementData[size++] = e;
        return true;
    }

    // ...
}

從中我們可以關(guān)注到關(guān)于 ArrayList 的幾點信息：

1. 使用數(shù)組存儲數(shù)據(jù)，即 elementData
2. 使用 int 成員變量 size 記錄實際元素個數(shù)
3. add 方法邏輯與執(zhí)行順序：
   • 執(zhí)行 ensureCapacityInternal(size + 1)：確認(rèn) elementData 的容量是否夠用，不夠用的話擴容一半（申請一個新的大數(shù)組，將 elementData 里的原有內(nèi)容 copy 過去，然后將新的大數(shù)組賦值給 elementData）
   • 執(zhí)行 elementData[size] = e;
   • 執(zhí)行 size++

為了方便理解這里討論的「線程安全問題」，我們選一個最簡單的執(zhí)行路徑來分析，假設(shè)有 A 和 B 兩個線程同時調(diào)用 ArrayList.add 方法，而此時 elementData 容量為 8，size 為 7，足以容納一個新增的元素，那么可能發(fā)生什么現(xiàn)象呢？

Thread Safety ArrayList Add

一種可能的執(zhí)行順序是：

• 線程 A 和 B 同時執(zhí)行了 ensureCapacityInternal(size + 1)，因 7 + 1 并沒超過 elementData 的容量 8，所以并未擴容
• 線程 A 先執(zhí)行 elementData[size++] = e;，此時 size 變?yōu)?8
• 線程 B 執(zhí)行 elementData[size++] = e;，因為 elementData 數(shù)組長度為 8，卻訪問 elementData[8]，數(shù)組下標(biāo)越界

程序會拋出異常，無法正常執(zhí)行完，根據(jù)前文提到的線程安全的定義，很顯然這已經(jīng)是屬于線程不安全的情況了。

構(gòu)造示例代碼驗證

有了以上的理解之后，我們來寫一段簡單的示例代碼，驗證以上問題確實可能發(fā)生：

List<Integer> resultList = new ArrayList<>();
List<Integer> paramList = new ArrayList<>();
int length = 10000;
for (int i = 0; i < length; i++) {
    paramList.add(i);
}
paramList.parallelStream().forEach(resultList::add);

執(zhí)行以上代碼有可能表現(xiàn)正常，但更可能是遇到以下異常：

Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
    at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:598)
    at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677)
    at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735)
    at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:174)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
    at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
    at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:583)
    at concurrent.ConcurrentTest.main(ConcurrentTest.java:18)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 1234
    at java.util.ArrayList.add(ArrayList.java:465)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
    at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
    at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)
    at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1067)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1703)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:172)

從我這里試驗的情況來看，length 值小的時候，因為達(dá)到容量邊緣需要擴容的次數(shù)少，不易重現(xiàn)，將 length 值調(diào)到比較大時，異常拋出率就很高了。

實際上除了拋出這種異常外，以上場景還可能造成數(shù)據(jù)覆蓋/丟失、ArrayList 里實際存放的元素個數(shù)與 size 值不符等其它問題，感興趣的同學(xué)可以繼續(xù)挖掘一下。

解決方案

對這類問題常見的有效解決思路就是對共享的資源訪問加鎖。

我提出代碼審查的修改意見后，小伙伴將文首代碼里的

List<String> resultList = new ArrayList<>();

修改為了

List<String> resultList = Collections.synchronizedList(new ArrayList<>());

這樣實際最終會使用 SynchronizedRandomAccessList，看它的實現(xiàn)類，其實里面也是加鎖，它內(nèi)部持有一個 List，用 synchronized 關(guān)鍵字控制對 List 的讀寫訪問，這是一種思路——使用線程安全的集合類，對應(yīng)的還可以使用 Vector 等其它類似的類來解決問題。

另外一種方思路是手動對關(guān)鍵代碼段加鎖，比如我們也可以將

resultList.add(value);

修改為

synchronized (mutex) {
    resultList.add(value);
}

小結(jié)

Java 8 的并行流提供了很方便的并行處理、提升程序執(zhí)行效率的寫法，我們在編碼的過程中，對用到多線程的地方要保持警惕，有意識地預(yù)防此類問題。

對應(yīng)的，我們在做代碼審查的過程中，也要對涉及到多線程使用的場景時刻繃著一根弦，在代碼合入前把好關(guān)，將隱患拒之門外。

參考

• 線程安全——維基百科