如何重新排序流的复杂未来？

Question

我处理可完成的未来流。这些需要不同的时间来完成。那些需要更长时间的块流处理，而其他人可能已经完成了（我知道平行流）

因此，我想在一个流中重新排序项目（例如，使用缓冲区），以向前移动已完成的期货。

例如，如果一个getUser调用需要很长时间，此代码将阻止流处理

public static Boolean isValid(User user) { ... }

emails.stream()
   // not using ::
   // getUser() returns CompletableFuture<User>
  .map( e -> getUser(e))
  // this line blocks Stream processing
  .filter( userF -> isValid( userF.get()) )
  .map( f -> f.thenApply(User::getName))

我想要一些

emails.stream()
   .map( e -> getUser(e))
   // this moves Futures into a bounded buffer
   // and puts those finished first
   // like CompletionService [1]
   // and returns a Stream again
   .collect(FutureReorderer.collector())
   // this is not the same Stream but
   // the one created by FutureReorderer.collector()
   .filter( userF -> isValid( userF.get()) )
   .map( f -> f.thenApply(User::getName))

[1] 例如CompletionServicehttps://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorCompletionService.html调用take（）时返回已完成的任务，否则返回块。但CompletionService不接受未来，需要做cs。萨姆比特-

我该怎么做？

[编辑]

1. 将示例更改为包括Filter（）
2. 添加评论
3. 添加CompletionService链接

Answer 1

我假设OP中的需求是同时执行getUser并按完成顺序处理结果Futures。以下是ExecutorCompletionService的解决方案：

final CompletionService<User> ecs = new ExecutorCompletionService<>(executor);

emails.stream().map(e -> ecs.submit(() -> getUser(e).get()))
    .collect(Collectors.collectingAndThen(Collectors.toList(), fs -> fs.stream())) // collect the future list for concurrent execution
    .map(f -> {
            try {
                return ecs.take().get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        })
    .filter(u -> isValid(u)).map(User::getName)... //TODO;

或：

final BlockingQueue<Future<User>> queue = new ArrayBlockingQueue<>(emails.size());
final CompletionService<User> ecs = new ExecutorCompletionService<>(executor, queue);        
emails.stream().forEach(e -> ecs.submit(() -> getUser(e).get()));

IntStream.range(0, emails.size())
    .mapToObj(i -> {
            try {
                return queue.poll().get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        })
    .filter(u -> isValid(u)).map(User::getName);

这很简单，但并不直接。

Answer 2

有更多的背景肯定会有助于调整答案——我觉得问题在其他地方，可以以更简单的方式解决。

但如果你的问题是如何在一开始就以某种方式保持完整的未来，那么几乎没有选择：

使用自定义的比较器对流进行排序：

.sorted(Comparator.comparing(f -> !f.isDone()))

请记住，isDone不仅在未来成功完成时返回true。

将期货存储在优先队列中

PriorityQueue<CompletableFuture<String>> queue
 = new PriorityQueue<>(Comparator.comparing(f -> !f.isDone()));

轮询元素时，队列将根据元素提供的顺序返回元素。

这就是行动：

PriorityQueue<CompletableFuture<String>> queue
 = new PriorityQueue<>(Comparator.comparing(f -> !f.isDone()));

queue.add(CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(Integer.MAX_VALUE);
    } catch (InterruptedException e) {  }

    return "42";
}));

queue.add(CompletableFuture.completedFuture("completed"));

queue.poll(); // "completed"
queue.poll(); // still going on

重要的是要记住，如果要将PriorityQueue转换为Stream，则不能简单地使用Stream（）——这将不会保留优先级顺序。

这是正确的选择：

Stream.generate(queue::poll).limit(queue.size())