java - 合并两个流

Question

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

我有办法做到这一点，我迭代流并将值插入 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，因为它是一个相当连续的操作。轻轻地使用 Guava ，

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