代碼審查：從 ArrayList 說線程安全

本文從代碼審查過程當中發現的一個 ArrayList 相關的「線程安全」問題出發，來剖析和理解線程安全。

案例分析

前兩天在代碼 Review 的過程當中，看到有小夥伴用了相似如下的寫法：

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

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

印象中 ArrayList 是線程不安全的，而這裏會多線程改寫同一個 ArrayList 對象，感受這樣的寫法會有問題，因而看了下 ArrayList 的實現來確認問題，同時複習下相關知識。

先貼個概念：

線程安全 是程式設計中的術語，指某個函數、函數庫在多線程環境中被調用時，可以正確地處理多個線程之間的共享變量，使程序功能正確完成。 ——維基百科

咱們來看下 ArrayList 源碼裏與本話題相關的關鍵信息：

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;
    }

    // ...
}

從中咱們能夠關注到關於 ArrayList 的幾點信息：

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

爲了方便理解這裏討論的「線程安全問題」，咱們選一個最簡單的執行路徑來分析，假設有 A 和 B 兩個線程同時調用 ArrayList.add 方法，而此時 elementData 容量爲 8，size 爲 7，足以容納一個新增的元素，那麼可能發生什麼現象呢？

一種可能的執行順序是：

• 線程 A 和 B 同時執行了 ensureCapacityInternal(size + 1)，因 7 + 1 並沒超過 elementData 的容量 8，因此並未擴容
• 線程 A 先執行 elementData[size++] = e;，此時 size 變爲 8
• 線程 B 執行 elementData[size++] = e;，由於 elementData 數組長度爲 8，卻訪問 elementData[8]，數組下標越界

程序會拋出異常，沒法正常執行完，根據前文提到的線程安全的定義，很顯然這已是屬於線程不安全的狀況了。

構造示例代碼驗證

有了以上的理解以後，咱們來寫一段簡單的示例代碼，驗證以上問題確實可能發生：

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);

執行以上代碼有可能表現正常，但更多是遇到如下異常：

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 值小的時候，由於達到容量邊緣須要擴容的次數少，不易重現，將 length 值調到比較大時，異常拋出率就很高了。

實際上除了拋出這種異常外，以上場景還可能形成數據覆蓋/丟失、ArrayList 裏實際存放的元素個數與 size 值不符等其它問題，感興趣的同窗能夠繼續挖掘一下。

解決方案

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

我提出代碼審查的修改意見後，小夥伴將文首代碼裏的

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

修改成了

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

這樣實際最終會使用 SynchronizedRandomAccessList，看它的實現類，其實裏面也是加鎖，它內部持有一個 List，用 synchronized 關鍵字控制對 List 的讀寫訪問，這是一種思路——使用線程安全的集合類，對應的還可使用 Vector 等其它相似的類來解決問題。

另一種方思路是手動對關鍵代碼段加鎖，好比咱們也能夠將

resultList.add(value);

修改成

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

小結

Java 8 的並行流提供了很方便的並行處理、提高程序執行效率的寫法，咱們在編碼的過程當中，對用到多線程的地方要保持警戒，有意識地預防此類問題。

對應的，咱們在作代碼審查的過程當中，也要對涉及到多線程使用的場景時刻繃着一根弦，在代碼合入前把好關，將隱患拒之門外。

參考

• 線程安全——維基百科