java - 制作流副本的最有效方法是什么？

Question

我有一个对流执行处理的方法。该处理的一部分需要在锁的控制下完成 - 一个用于处理所有元素的锁定部分 - 但其中一些不需要(并且不应该因为它可能非常耗时)。所以我不能只说:

Stream<V> preprocessed = Stream.of(objects).map(this::preProcess);
Stream<V> toPostProcess;
synchronized (lockObj) {
    toPostProcess = preprocessed.map(this::doLockedProcessing);
}
toPostProcess.map(this::postProcess).forEach(System.out::println);

因为对 doLockedProcessing 的调用只会在终端操作 forEach 被调用时执行，而这在锁之外。

所以我认为我需要在每个阶段使用终端操作制作流的副本，以便在正确的时间完成正确的位。像这样的东西:

Stream<V> preprocessed = Stream.of(objects).map(this::preProcess).copy();
Stream<V> toPostProcess;
synchronized (lockObj) {
    toPostProcess = preprocessed.map(this::doLockedProcessing).copy();
}
toPostProcess.map(this::postProcess).forEach(System.out::println);

当然，copy() 方法不存在，但如果存在，它将对流执行终端操作并返回包含所有相同元素的新流。

我知道实现此目的的几种方法:

(1) 通过数组(如果元素类型是泛型类型就不那么容易了):

copy = Stream.of(stream.toArray(String[]::new));

(2) 通过列表:

copy = stream.collect(Collectors.toList()).stream();

(3) 通过流生成器:

Stream.Builder<V> builder = Stream.builder();
stream.forEach(builder);
copy = builder.build();

我想知道的是:这些方法中哪种方法在时间和内存方面最有效？或者还有其他更好的方法吗？

Answer 1

我想您已经提到了所有可能的选择。没有其他结构性方法可以满足您的需求。首先，您必须使用原始流。然后，创建一个新流，获取锁并使用这个新流(从而调用您的锁定操作)。最后，创建一个更新的流，释放锁并继续处理这个更新的流。

从您正在考虑的所有选项中，我会使用第三个，因为它可以处理的元素数量仅受内存限制，这意味着它没有隐式最大大小限制，例如 ArrayList 有(它可以包含大约 Integer.MAX_VALUE 个元素)。

不用说，这将是一个相当昂贵的操作，无论是在时间上还是在空间上。你可以这样做:

Stream<V> temp = Stream.of(objects)
        .map(this::preProcess)
        .collect(Stream::<V>builder,
                 Stream.Builder::accept,
                 (b1, b2) -> b2.build().forEach(b1))
        .build();

synchronized (lockObj) {
    temp = temp
            .map(this::doLockedProcessing)
            .collect(Stream::<V>builder,
                     Stream.Builder::accept,
                     (b1, b2) -> b2.build().forEach(b1))
            .build();
}

temp.map(this::postProcess).forEach(System.out::println);

请注意，我使用了单个 Stream 实例 temp，以便在需要时可以对中间流(及其构建器)进行垃圾回收。

---

正如@Eugene 在评论中所建议的那样，最好有一个实用方法来避免代码重复。这是这样的方法:

public static <T> Stream<T> copy(Stream<T> source) {
    return source.collect(Stream::<T>builder,
                          Stream.Builder::accept,
                          (b1, b2) -> b2.build().forEach(b1))
                 .build();
}

然后，您可以按如下方式使用此方法:

Stream<V> temp = copy(Stream.of(objects).map(this::preProcess));

synchronized (lockObj) {
    temp = copy(temp.map(this::doLockedProcessing));
}

temp.map(this::postProcess).forEach(System.out::println);