合并两个流

Question

我试图实现一个方法来合并两个Streams中的值基于a Comparator的值.

我有办法做到这一点,我迭代流并将值插入到一个Stream.Builder,但我还没有弄清楚如何制作一个懒惰评估版本(许多流操作的方式),所以它可以处理无限流.

我想要它做的就是对输入数据执行单个合并传递,而不是对流进行排序(实际上,流很可能是无序的;需要保留这种混乱).

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

我如何懒惰地合并这样的两个流？

如果我用两个Queues作为输入,一些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 */ );