Java8-parallelStream()使用

liuian 2025-01-09 14:25 46 浏览

parallelStream()

parallelStream() 是 Java 8 引入的一种方法，属于 Stream 接口的一部分。它用于提供并行处理流数据的能力，从而利用多核处理器提高性能。通过使用 parallelStream()，可以轻松实现数据的并行处理，而无需手动管理线程或并发。

基本概念

Stream: 是一种用于处理集合的抽象，可以对集合数据进行过滤、映射、排序等操作。
parallelStream(): 该方法返回一个 **并行流**，允许流中的元素被分配到多个线程中进行并行处理。

如何使用 parallelStream()

以下是使用 parallelStream() 的基本步骤和示例：

创建集合：首先，您需要一个集合（如 List、Set 等）。
调用 parallelStream()：在集合上调用 parallelStream() 方法。
使用流操作：对流进行各种操作，如 map、filter、collect 等。
终止操作：使用终止操作（如 forEach、collect）来执行流的操作。

示例代码

import java.util.Arrays;
import java.util.List;

public class ParallelStreamExample {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Alice", "Bob", "Charlie", "David", "Eve");

        // 使用 parallelStream() 进行并行处理
        names.parallelStream()
             .filter(name -> name.length() > 3) // 过滤出长度大于 3 的名字
             .map(String::toUpperCase) // 将名字转换为大写
             .forEach(System.out::println); // 打印结果
    }
}

import java.util.concurrent.ForkJoinPool;

public class CustomParallelStream {
    public static void main(String[] args) {
        ForkJoinPool customThreadPool = new ForkJoinPool(4); // 创建一个具有4个线程的线程池

        customThreadPool.submit(() -> {
            List<String> names = Arrays.asList("Alice", "Bob", "Charlie", "David", "Eve");
            names.parallelStream()
                 .forEach(name -> System.out.println(name + " processed by " + Thread.currentThread().getName()));
        }).join(); // 等待任务完成
    }
}

注意事项

性能: 在某些情况下，使用 parallelStream() 可以显著提高性能，尤其是在处理大量数据时。然而，对于小规模数据集，开启并行处理可能会导致性能下降，因为线程管理的开销可能超过并行处理带来的好处。
如果并行里面存在查询数据库可能对数据库造成成倍压力，需要做好一定预估。
无状态操作: 并行流中的操作是无状态的，确保在处理过程中不会引入共享可变状态。这意味着如果你的操作依赖于某个共享的变量，可能会导致数据不一致和错误。
线程安全: 确保在使用并行流时，任何共享的可变对象都是线程安全的。例如，使用 ConcurrentHashMap 或其他线程安全的集合类来存储结果。
调试: 调试并行流可能会比较困难，因为流中的操作可能在不同的线程中异步执行。

使用场景

假设有一个订单包装返回值的逻辑，订单需要处理调用物流信息、订单基础信息、商品基础信息、退货信息等，目前耗时比较高可能达到400ms了，这时你可以：创建一个IBuildHandler 接口，包含 handler 方法，如下所示：

public interface IBuildHandler {void handler(Condition condition, List<UserOrder> userOrderList);
}

然后，你可以使用并行流来处理多个处理器进行处理逻辑：

List<IBuildHandler> buildHandlers = Arrays.asList(new BuildHandlerA(), new BuildHandlerB(), new BuildHandlerC());
Request req = new Request();
List<UserOrder> userOrderList = getUserOrderList();
buildHandlers.parallelStream()
    .forEach(iBuildHandler -> iBuildHandler.handler(req.getCondition(), userOrderList));

工作原理

分割任务: 当你调用 parallelStream() 时，Java 会将原始数据集合分割成多个小块（也称为任务）。这些小块将被分配给多个线程进行处理。
Fork/Join 框架: parallelStream() 底层使用了 Java 的 Fork/Join 框架。这个框架专门用于实现并行处理，能够有效地管理任务的分割和合并。
线程池: 默认情况下，Fork/Join 框架使用一个公共的 ForkJoinPool，它的线程数量通常是 CPU 核心数的两倍。每个任务会在这个线程池中执行。