java - 如何在可完成的 future 流中压平列表？

Question

我有这个:

Stream<CompletableFuture<List<Item>>>

如何转换成

Stream<CompletableFuture<Item>>

其中:第二个流由第一个流中每个列表中的每个和所有项目组成。

我查看了 thenCompose 但它解决了一个完全不同的问题，也称为“扁平化”。

如何以流方式高效地完成这项工作，而不阻塞或过早地消耗不必要的更多流项目？

这是我迄今为止最好的尝试:

    ExecutorService pool = Executors.newFixedThreadPool(PARALLELISM);
    Stream<CompletableFuture<List<IncomingItem>>> reload = ... ;

    @SuppressWarnings("unchecked")
    CompletableFuture<List<IncomingItem>> allFutures[] = reload.toArray(CompletableFuture[]::new);
    CompletionService<List<IncomingItem>> queue = new ExecutorCompletionService<>(pool);
    for(CompletableFuture<List<IncomingItem>> item: allFutures) {
        queue.submit(item::get);
    }
    List<IncomingItem> THE_END = new ArrayList<IncomingItem>();
    CompletableFuture<List<IncomingItem>> ender = CompletableFuture.allOf(allFutures).thenApply(whatever -> {
        queue.submit(() -> THE_END);
        return THE_END;
    });
    queue.submit(() -> ender.get());
    Iterable<List<IncomingItem>> iter = () -> new Iterator<List<IncomingItem>>() {
        boolean checkNext = true;
        List<IncomingItem> next = null;
        @Override
        public boolean hasNext() {
            if(checkNext) {
                try {
                    next = queue.take().get();
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
                checkNext = false;
            }
            if(next == THE_END || next == null) {
                return false;
            }
            else {
                return true;
            }
        }
        @Override
        public List<IncomingItem> next() {
            if(checkNext) {
                hasNext();
            }
            if(!hasNext()) {
                throw new IllegalStateException();
            }
            checkNext = true;
            return next;
        }
    };
    Stream<IncomingItem> flat = StreamSupport.stream(iter.spliterator(), false).flatMap(List::stream);

这起初有效，不幸的是，它有一个 fatal error :结果流似乎在检索所有项目之前过早终止。

Answer 1

正如我在评论中所写，这是不可能的。

考虑一些任意服务，它将返回 CompletableFuture<Integer> :

CompletableFuture<Integer> getDiceRoll();

我现在可以转换这个 CompletableFuture<Integer>到 Stream<CompletableFuture<List<Object>>>没有任何问题:

Stream<CompletableFuture<List<Object>>> futureList = Stream.of(getDiceRoll().thenApply(n -> List.of(new Object[n])));

让我们假设有一个通用的方法来转换 Stream<CompletableFuture<List<T>>>进入 Stream<CompletableFuture<T>> :

<T> Stream<CompletableFuture<T> magic(Stream<CompletableFuture<List<T>>> arg);

然后我可以执行以下操作:

int diceRoll = magic(Stream.of(getDiceRoll().thenApply(n -> List.of(new Object[n])))).count();

等等，什么？
我现在可以从 CompletableFuture 中得到一个任意整数.这意味着，通过一些工程努力，我可以从 CompletableFuture 中获取所有信息 - 毕竟，内存只是一些数字。

所以我们必须得出结论，像magic这样的方法不违反时间结构就不可能存在。
这就是答案:没有这样的方法，因为它不可能存在。