Java 8 引入了強(qiáng)大的 Stream API，為處理集合數(shù)據(jù)提供了簡潔、高效的解決方案。其中，parallel() 方法為流處理引入了并行化能力，允許開發(fā)者充分利用多核處理器的優(yōu)勢，大幅提升大規(guī)模數(shù)據(jù)集的處理效率。

本篇文章將帶你開啟并行流處理之旅，認(rèn)識 Java 8 Stream API 中的 parallel()。

什么是 parallel()

parallel() 是 Java 8 Stream API 中的一個(gè)方法，用于將一個(gè)順序流轉(zhuǎn)換為并行流。并行流是一種可以同時(shí)在多個(gè)線程上執(zhí)行操作的流，它將流的元素分割成多個(gè)子集，每個(gè)子集在不同的線程上獨(dú)立處理，最后將結(jié)果合并。使用 parallel() 方法可以輕松開啟并行流處理模式，無需顯式管理線程和同步。

List<Integer> numbers = ...; // 假設(shè)有一個(gè)包含大量元素的列表

numbers.stream() // 創(chuàng)建順序流
    .parallel() // 轉(zhuǎn)換為并行流
    .filter(n -> n % 2 == 0) // 并行過濾偶數(shù)
    .map(n -> n * 2) // 并行映射為原數(shù)的兩倍
    .forEach(System.out::println); // 并行打印結(jié)果

在這個(gè)示例中，parallel() 方法將順序流轉(zhuǎn)換為并行流，后續(xù)的 filter()、map() 和 forEach() 操作將在多個(gè)線程上并行執(zhí)行，從而加速數(shù)據(jù)處理。

并行流的工作原理

并行流處理背后的核心機(jī)制主要包括以下幾個(gè)方面：

1. 分割與合并
2. 自動(dòng)流水線化
3. 適應(yīng)性執(zhí)行策略

并行流根據(jù)數(shù)據(jù)集的大小、處理器核心數(shù)等因素動(dòng)態(tài)調(diào)整并行度和任務(wù)劃分策略。對于小規(guī)模數(shù)據(jù)集或不適合并行化的操作，Java 8 會(huì)自動(dòng)退化為順序流處理，避免不必要的線程開銷。

總之，parallel() 方法通過將原始列表拆分成多個(gè)子任務(wù)，并在獨(dú)立線程上并行執(zhí)行流操作鏈的各個(gè)階段，最后合并處理結(jié)果，實(shí)現(xiàn)了對列表數(shù)據(jù)的高效并行處理。具體的拆分策略和并行執(zhí)行細(xì)節(jié)由 JVM 自動(dòng)管理，開發(fā)者無需關(guān)心底層實(shí)現(xiàn)，只需關(guān)注流式編程的高層抽象。

實(shí)戰(zhàn)應(yīng)用

適合parallel()并行流的應(yīng)用場景有：

1. 大規(guī)模數(shù)據(jù)集處理
2. CPU 密集型操作
3. 可并行化的中間操作，如 filter()、map()、flatMap()、sorted()等。

示例1：大規(guī)模數(shù)據(jù)集處理

場景：在一個(gè)數(shù)據(jù)分析項(xiàng)目中，需要對一個(gè)包含百萬條記錄的數(shù)據(jù)集進(jìn)行復(fù)雜過濾和計(jì)算。使用并行流可以顯著加快處理速度，充分利用多核處理器資源。示例

public class ParallelDataProcessingExample {
    public static void main(String[] args) {
        List<DataRecord> records = generateLargeDataRecords(); // 假設(shè)生成包含百萬條記錄的數(shù)據(jù)集

        List<DataRecord> filteredAndProcessedRecords = records.parallelStream()
                .filter(record -> record.isValid()) // 并行過濾有效記錄
                .map(record -> record.computeComplexMetric()) // 并行計(jì)算復(fù)雜度量
                .collect(Collectors.toList());

        // ... 使用 filteredAndProcessedRecords 進(jìn)行后續(xù)分析 ...
    }

}

public class DataRecord {
    // ... 數(shù)據(jù)記錄的字段、方法等 ...

    public boolean isValid() {
        // ... 判斷記錄是否有效的邏輯 ...
    }

    public DataRecord computeComplexMetric() {
        // ... 計(jì)算復(fù)雜度量的邏輯 ...
    }
}

示例2

場景：假設(shè)有一個(gè)電商系統(tǒng)需要批量更新大量商品的價(jià)格，每個(gè)商品的更新過程涉及網(wǎng)絡(luò)請求到不同服務(wù)獲取最新價(jià)格信息，然后保存到數(shù)據(jù)庫。

示例：

@Service
@RequiredArgsConstructor
public class ProductService {

    private final PriceService priceService;
    private final ProductRepository productRepository;
    private final Executor asyncExecutor;


    /**
  * 批量更新商品價(jià)格
  *
  * @param productIds 商品ID列表
  */
 public void batchUpdatePrices(List<Integer> productIds) {
  CompletableFuture<Void> allDbUpdates = CompletableFuture.allOf(productIds.stream()
    .parallel()
    .map(productId -> CompletableFuture.supplyAsync(() -> priceService.getLatestPrice(productId), asyncExecutor)
      .thenAcceptAsync(newPrice -> productRepository.updatePrice(productId, newPrice), asyncExecutor))
    .toArray(CompletableFuture[]::new));

  // 等待所有數(shù)據(jù)庫更新完成
  allDbUpdates.join();
 }
}

在這個(gè)示例中：

• 首先，我們創(chuàng)建了一個(gè)包含100個(gè)商品ID的列表，并對其應(yīng)用了 parallel() 流操作，使得后續(xù)的 map() 操作能并行執(zhí)行。
• 為每個(gè)商品ID創(chuàng)建一個(gè) CompletableFuture，通過 supplyAsync() 異步調(diào)用 PriceService 獲取最新價(jià)格。
• 進(jìn)一步使用 thenAcceptAsync() 異步操作。在獲取到最新價(jià)格之后更新數(shù)據(jù)庫。
• 最終，使用 CompletableFuture.allOf() 等待所有數(shù)據(jù)庫更新操作完成。

小結(jié)

Java 8 Stream API 中的 parallel() 方法為處理集合數(shù)據(jù)提供了便捷的并行化途徑。

在復(fù)雜的異步處理場景中，可以結(jié)合 CompletableFuture 與并行流，進(jìn)一步提升程序的并發(fā)性和響應(yīng)能力。通過合理使用并行流，開發(fā)者可以顯著提升大規(guī)模數(shù)據(jù)集處理的性能，充分發(fā)揮現(xiàn)代多核處理器的潛力。

然而，使用并行流時(shí)也應(yīng)注意避免數(shù)據(jù)依賴、狀態(tài)共享等問題，適時(shí)進(jìn)行性能評估與調(diào)整。