Hợp nhất hai luồng

Question

Tôi đang cố gắng triển khai phương pháp hợp nhất các giá trị trong hai số Stream s dựa trên giá trị Comparator cho các giá trị.

Tôi có cách để làm điều này, nơi tôi lặp qua các luồng và chèn các giá trị vào một Stream.Builder, nhưng tôi đã không thể tìm ra cách tạo phiên bản được đánh giá lười biếng (cách hoạt động của nhiều luồng), vì vậy nó cũng có thể xử lý các luồng vô hạn.

Tất cả tôi muốn nó phải làm là thực hiện một đơn sáp nhập vượt qua trên các dữ liệu đầu vào, không loại các dòng (trên thực tế, có khả năng là các con suối sẽ bị rối loạn; rối loạn này cần phải được bảo quản) .

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

Làm cách nào để tôi có thể hợp nhất hai luồng như thế này?

Nếu tôi được làm điều này với hai Queue s như là đầu vào và một số Consumer như đầu ra, nó sẽ là khá đơn giản:

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); 
} 

Nhưng tôi cần phải làm điều này với đánh giá lười biếng, và suối.

Để giải quyết những ý kiến, đây là một số nguyên liệu đầu vào ví dụ và kết quả:

Ví dụ 1:

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

sẽ trả về một dòng mà sẽ tạo ra chuỗi này:

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

Ví dụ 2:

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

sẽ trả về một vô hạn (tốt, một mục INT_MAX + 5) dòng đó sẽ tạo ra chuỗi:

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

Như bạn có thể thấy, đây không phải là chỉ đơn thuần concat(first,second).sort(), vì (a) bạn không thể loại suối vô hạn, và (b) ngay cả khi bạn có thể sắp xếp các luồng, nó không đưa ra kết quả mong muốn.

Answer 1

Bạn cần triển khai Spliterator, thay vì xem qua Stream.Builder. Đối với điều này, bạn thậm chí có thể chỉ cần đi qua một Iterator, vì nó là một hoạt động khá tuần tự. Sử dụng ổi nhẹ,

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