java - 从迭代器创建的 CompletableFuture 流不会延迟计算

Question

我对可完成 future 的完成方式和时间有些纠结。我创建了这个测试用例:

import org.junit.Test;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class StreamOfCompletableFuturesTest {
    @Test
    public void testList() {
        completeFirstTwoElements(
                Stream.of("list one", "list two", "list three", "list four", "list five")
        );
    }

    @Test
    public void testIterator() {
        Iterator<String> iterator = Arrays.asList("iterator one", "iterator two", "iterator three", "iterator four", "iterator five").iterator();

        completeFirstTwoElements(
            StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false)
        );
    }

    private void completeFirstTwoElements(Stream<String> stream) {
        stream
                .map(this::cf)
                .limit(2)
                .parallel()
                .forEach(cf -> {
                    try {
                        System.out.println(cf.get());
                    } catch (InterruptedException | ExecutionException e) {
                        throw new RuntimeException(e);
                    }
                });
    }

    private CompletableFuture<String> cf(String result) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("Running " + result);
            return result;
        });
    }
}

输出是:

Running list one
Running list two
list two
list one
Running iterator one
Running iterator two
Running iterator three
Running iterator four
Running iterator five
iterator two
iterator one

testList 方法按预期工作。 CompletableFuture 仅在最后评估，因此在 limit 方法仅保留前两项之后。

然而，testIterator 方法却出乎意料。所有 CompletableFuture 都已完成，限制仅在之后完成。

如果我从流中删除 parallel() 方法，它将按预期工作。但是，处理(forEach())应该并行完成，因为在我的完整程序中它是一个长时间运行的方法。

谁能解释为什么会这样？

看起来这取决于Java版本，所以我在1.8:

$ java -version
java version "1.8.0_92"
Java(TM) SE Runtime Environment (build 1.8.0_92-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.92-b14, mixed mode)

Answer 1

并行适用于整个管道，因此您无法真正控制在 limit() 应用于并行 Stream 之前将执行什么。唯一的保证是 limit() 之后的内容只会在保留的元素上执行。

两者之间的差异可能是由于一些实现细节或其他 Stream 特性。事实上，您可以通过使用 SIZED 特征轻松反转行为。似乎当 Stream 的大小已知时，只处理了 2 个元素。

例如，应用一个简单的 filter() 将丢失列表版本的大小:

completeFirstTwoElements(
        Stream.of("list one", "list two", "list three", "list four", "list five").filter(a -> true)
);

例如输出:

Running list one
Running list five
Running list two
Running list three
list one
list two

并且不使用 未知大小 版本的 Spliterator.spliterator() “修复”了行为:

Iterator<String> iterator = Arrays.asList("iterator one", "iterator two", "iterator three", "iterator four", "iterator five").iterator();

completeFirstTwoElements(
        StreamSupport.stream(Spliterators.spliterator(iterator, Spliterator.ORDERED, 5), false)
);

输出:

Running iterator two
Running iterator one
iterator one
iterator two