詳解ParallelStream

一、ParallelStream.foreach

ParallelStream.foreach方法可以遍歷一個Stream中的所有元素，並對每個元素執行一些操作。但與普通的Stream不同的是，ParallelStream可以使用並行計算的方式進行操作，從而提高效率。

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
list.parallelStream().forEach(i -> {
    System.out.println(Thread.currentThread().getName() + ": " + i);
});

代碼中的list.parallelStream()創建了一個ParallelStream，使用forEach方法遍歷每個元素，打印出當前執行操作的線程名和元素值。

如果使用普通的Stream，代碼如下：

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
list.stream().forEach(i -> {
    System.out.println(Thread.currentThread().getName() + ": " + i);
});

可以看出，兩段代碼的輸出結果是一樣的，但使用ParallelStream可以比普通Stream更快地執行。

二、ParallelStream的工作原理

ParallelStream使用分割-執行-合併的方式進行操作。簡單來說，就是將一個大任務分割成小任務，交給不同的線程去執行，最後將所有結果合併。

首先，輸入的元素會被分割成多個小任務，每個小任務由一個線程去執行。這個過程可以通過以下代碼來演示：

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
list.parallelStream().peek(i -> System.out.println(Thread.currentThread().getName() + ": " + i)).count();

代碼中的peek方法可以在每個元素被處理之前輸出當前執行操作的線程名和元素值。count方法用於觸發操作。

對於上述代碼的輸出結果，可以看到不同的元素由不同的線程處理：

ForkJoinPool.commonPool-worker-1: 4
ForkJoinPool.commonPool-worker-1: 3
ForkJoinPool.commonPool-worker-1: 1
ForkJoinPool.commonPool-worker-2: 6
ForkJoinPool.commonPool-worker-2: 5
ForkJoinPool.commonPool-worker-2: 2
ForkJoinPool.commonPool-worker-3: 10
ForkJoinPool.commonPool-worker-3: 9
ForkJoinPool.commonPool-worker-3: 7
ForkJoinPool.commonPool-worker-4: 8

在執行完所有小任務之後，所有結果會被合併。合併的方式是根據操作類型來確定的。例如，對於求和操作，所有小任務的結果會被加起來，以得到最終的結果。

三、ParallelStream的注意事項

1. 數據源必須是線程安全的

由於ParallelStream將一個任務分割成多個小任務，並可能由不同的線程執行，因此數據源必須是線程安全的。否則會在多個線程並發執行的過程中出現競態條件，並可能導致不確定的結果。

例如，下面的代碼會導致ConcurrentModificationException異常：

List<Integer> list = new ArrayList<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
list.parallelStream().forEach(i -> {
    if (i == 5) {
        list.remove(i);
    }
});

因為在刪除元素的過程中會導致並發修改數據源，而數據源是ArrayList，不是線程安全的。

如果使用線程安全的數據源，上述代碼就不會出現異常：

List<Integer> list = new CopyOnWriteArrayList<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
list.parallelStream().forEach(i -> {
    if (i == 5) {
        list.remove(i);
    }
});

2. 對於有狀態的操作，需要保證線程安全

當對一個Stream進行有狀態的操作時，需要保證線程安全。

有狀態的操作包括對變量的賦值、累加等操作，這些操作可能會導致線程安全問題。

例如，下面的代碼計算一個Stream中所有元素的和：

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
int sum = list.parallelStream().mapToInt(Integer::intValue).sum();
System.out.println(sum);

上述代碼輸出的結果是55，即所有元素的和。但如果在mapToInt方法中使用了一個全局變量，就會導致線程安全問題：

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
int count = 0;
int sum = list.parallelStream().mapToInt(i -> {
    count++;
    return i;
}).sum();
System.out.println(sum + " " + count);

輸出結果是不確定的，因為count變量被多個線程共享，並可能導致競態條件。

3. 避免使用過多的線程

使用ParallelStream時，需要確保使用恰當的線程數。過多的線程可能會導致線程上下文切換的開銷，反而影響效率。

可以通過以下方式來指定線程數：

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "4");

上述代碼指定了使用4個線程。

4. 記得關閉Stream

Stream有可能會佔用大量的內存，因此在使用完Stream後需要關閉它們。

可以通過以下方式來關閉一個Stream：

stream.close();

四、總結

在本文中，我們詳細闡述了ParallelStream的使用方法和工作原理，並提出了需要注意的事項。ParallelStream可以使得對Stream的操作並行化，從而提高效率。但在使用ParallelStream時，需要確保數據源是線程安全的，有狀態的操作需要保證線程安全，避免使用過多的線程，以及記得關閉Stream。