前言

最近大聰明一直在開(kāi)發(fā)項(xiàng)目中的新需求，其中有一個(gè)需求是“解析文件（.txt文件，一行就是一條數(shù)據(jù)）中的數(shù)據(jù)并進(jìn)行入庫(kù)操作”。其實(shí)這個(gè)需求也很簡(jiǎn)單，無(wú)非就是將文件中每一行數(shù)據(jù)轉(zhuǎn)換為一個(gè)對(duì)象，將每一個(gè)對(duì)象都存儲(chǔ)到 list 集合中，最終執(zhí)行批量入庫(kù)的操作。但就是這么一個(gè)簡(jiǎn)單的需求卻讓我踩了一個(gè)大坑....

踩坑日記

各位小伙伴先看一下上圖中的代碼，不知道各位小伙伴有沒(méi)有看出什么問(wèn)題呢？?? 可能這么看起來(lái)有些不好理解，咱們?cè)俸?jiǎn)化一下圖中的代碼，如下所示：??

public static void main(String[] args) {
        List<Integer> list = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            list.add(i);
        }
        System.out.println("a:"+list.size());
        List<Integer> streamList = new ArrayList<>();
        list.parallelStream().forEach(streamList::add);
        System.out.println("b:"+streamList.size());
    }
}

各位小伙伴看看簡(jiǎn)化后的代碼，猜測(cè)一下 a 和 b 輸出的值分別是多少呢？這里就不賣(mài)關(guān)子了，咱們直接揭曉答案 ??

結(jié)果可能和大多數(shù)小伙伴猜測(cè)的都不太一樣，a 和 b 的值居然不相等，且 b 的值 永遠(yuǎn)都會(huì)小于 a，同時(shí)在多次執(zhí)行之后可能會(huì)出現(xiàn)數(shù)組下標(biāo)越界異常，顯然這里的代碼是不符合邏輯的??

這也是我在項(xiàng)目中遇倒的問(wèn)題所在，解析完文件后，通過(guò) parallelStream().forEach() 遍歷結(jié)果進(jìn)行處理，但是最終入庫(kù)的數(shù)據(jù)條數(shù)總是小于文件中的數(shù)據(jù)條數(shù)。

這種情況大聰明還是第一次見(jiàn)到，不過(guò)卻又一次激起了大聰明的求知欲，開(kāi)啟了刨根問(wèn)底模式~

刨根問(wèn)底

經(jīng)過(guò)大聰明的一番探索，也是終于找到了問(wèn)題答案... ??

Stream（流）是 JDK8 中引入的一種類(lèi)似與迭代器（Iterator）的單向迭代訪問(wèn)數(shù)據(jù)的工具。ParallelStream 則是并行的流，它通過(guò) Fork/Join 框架來(lái)拆分任務(wù)，加速流的處理過(guò)程。Fork/Join 的框架是通過(guò)把一個(gè)大任務(wù)不斷 fork 成許多子任務(wù)，然后多線程執(zhí)行這些子任務(wù)，最后再 Join 這些子任務(wù)得到最終結(jié)果。咱們回到實(shí)例代碼中來(lái)解釋一下，就是先將 list 集合 fork 成多段，然后多線程添加到 streamList 的結(jié)合中，而 streamList 是ArrayList 類(lèi)型，ArrayList 的 add() 方法并不能保證原子性。

咱們先看一下 ArrayList 中 add() 方法的源碼??

眾所周知，ArrayList 作為 Collection 中極重要的一員，是非線程安全的，所以 ArrayList 并不適合多線程高并發(fā)的情況，在多線程高并發(fā)時(shí)會(huì)出現(xiàn)內(nèi)部某些位置為 null 的情況。核心原因是，ArrayList 的add() 的方法不是線程安全的，是非原子性的，add操作可以簡(jiǎn)單理解為兩個(gè)步驟：

• ensureCapacityInternal(size + 1) ：確認(rèn)當(dāng)前 ArrayList 中的數(shù)組是否還可以加入新的元素。如果不行，就會(huì)再申請(qǐng)一個(gè)：int newCapacity = oldCapacity + (oldCapacity >> 1) 大小的數(shù)組（即容量變?yōu)樵瓉?lái)的 1.5 倍），然后將數(shù)據(jù)復(fù)制過(guò)去。
• elementData[size++] = e：將元素添加到 elementData 數(shù)組中。

那么在多線程高并發(fā)情況下，如果有A、B兩個(gè)線程同時(shí)執(zhí)行 add() 方法，在第一步校驗(yàn)數(shù)組容量時(shí)，A、B線程都發(fā)現(xiàn)當(dāng)前無(wú)需擴(kuò)容，還可以繼續(xù)添加一個(gè)元素；因此A、B線程都進(jìn)入了第二步，此時(shí)，A線程先執(zhí)行完，數(shù)組容量已滿(mǎn)，然后B線程再對(duì) elementData 賦值時(shí)，就會(huì)出現(xiàn)我們上面說(shuō)到的情況，要么是數(shù)據(jù)丟失，要么是拋出數(shù)組下標(biāo)越界的異常。

解決方案

問(wèn)題原因我們已經(jīng)找到了，那么問(wèn)題的解決方案也就呼之欲出了~ ??

• ?? 方案一：將 parallelStream 改成 stream，或者直接使用 foreach 遍歷處理。也就是放棄多線程的寫(xiě)法，改為傳統(tǒng)的單線程處理。
• ?? 方案二：使用 list = new CopyOnWriteArrayList<>(); 這是個(gè)線程安全的類(lèi)。從源碼上看，CopyOnWriteArrayList 在 add 操作時(shí)，通過(guò) ReentrantLock 進(jìn)行加鎖，防止并發(fā)寫(xiě)。但是每次 add 操作都是把原數(shù)組中的元素拷貝一份到新數(shù)組中，然后在新數(shù)組中添加新元素，最后再把引用指向新數(shù)組，這也就會(huì)頻繁的創(chuàng)建數(shù)組（千萬(wàn)別忘了數(shù)組需要一塊連續(xù)的內(nèi)存空間）。所以當(dāng)實(shí)際業(yè)務(wù)邏輯中存在大量 add 操作時(shí)，要謹(jǐn)慎使用 CopyOnWriteArrayList 。
• ?? 方案三：使用包裝類(lèi) list = Collections.synchronizedList(Arrays.asList());

我們?cè)谑褂?parallelStream 之前，一定要仔細(xì)思考一下自己的業(yè)務(wù)邏輯是否真的需要多線程并發(fā)處理。其實(shí)在實(shí)際應(yīng)用場(chǎng)景中，并不是所有的問(wèn)題都適合使用并發(fā)來(lái)解決，比如當(dāng)數(shù)據(jù)量不大時(shí)，順序執(zhí)行往往比并行執(zhí)行更快，畢竟準(zhǔn)備線程池和其它相關(guān)資源也是需要時(shí)間的。但是，當(dāng)任務(wù)涉及到 I/O 操作并且任務(wù)之間不互相依賴(lài)時(shí)，那么并行化就是一個(gè)不錯(cuò)的選擇。

小結(jié)

到此這篇關(guān)于Java中parallelStream().forEach()的踩坑日記的文章就介紹到這了,更多相關(guān)Java parallelStream().forEach()內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論