合并两个流

Question

我正在尝试实现一个方法，根据值的一个Stream来合并两个Comparator中的值。

我有办法做到这一点，我在流中迭代并将值插入到Stream.Builder中，但我一直无法弄清楚如何生成一个延迟计算的版本(许多流操作都是这样)，因此它也可以处理无限流。

我想要做的就是对输入数据执行一次合并传递，而不是对流进行排序(实际上，流很可能会被乱排序；这种混乱需要保留)。

static Stream<E> merge(Stream<E> first, Stream<E> second, Comparator<E> c)

我怎么能像这样懒洋洋地合并两个流呢？

如果我使用两个Queue作为输入，一些Consumer作为输出，这将非常简单：

void merge(Queue<E> first, Queue<E> second, Consumer<E> out, Comparator<E> c){
    while(!first.isEmpty() && !second.isEmpty()
        if(c.compare(first.peek(), second.peek()) <= 0)
            out.accept(first.remove());
        else
            out.accept(second.remove());
    for(E e:first)
        out.accept(e);
    for(E e:second)
        out.accept(e);
}

但是我需要用懒惰的评估和流来做这件事。

为了处理这些注释，下面是一些示例输入和结果：

示例1：

merge(
    Stream.of(1, 2, 3, 1, 2, 3),
    Stream.of(2, 2, 3, 2, 2, 2),
    Comparator.naturalOrder()
);

将返回生成此序列的流：

1, 2, 2, 2, 3, 3, 1, 2, 2, 2, 2, 3

示例2：

merge(
    Stream.iterate(5, i->i-1),
    Stream.iterate(1, i->i+1),
    Comparator.naturalOrder()
);

将返回一个产生序列的无限流(嗯，一个INT_MAX + 5项)：

1, 2, 3, 4, 5, 5, 4, 3, 2, 1, 0, -1 ...

正如您所看到的，这不仅仅是concat(first,second).sort()，因为(a)您不能对无限流进行排序，并且(b)即使您可以对流进行排序，它也不会给出所需的结果。

Answer 1

您需要实现Spliterator，而不是通过Stream.Builder。为此，您甚至可能只需要通过一个Iterator，因为它是一个相当连续的操作。轻轻地用番石榴，

return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
    Iterators.mergeSorted(
      Arrays.asList(stream1.iterator(), stream2.iterator()),
      comparator),
    Spliterator.ORDERED),
  false /* not parallel */ );

Answer 2

Iterables.mergeSorted()来自番石榴

public static <T> Iterable<T> mergeSorted(Iterable<? extends Iterable<? extends T>> iterables,
                Comparator<? super T> comparator)

• https://javabot.evanchooly.com/javadoc/guava/22.0/com/google/common/collect/Iterables.html#mergeSorted-java.lang.Iterable-java.util.Comparator-