Strumień <Stream>: flatMap porównaniu zmniejszenia

Question

Jeśli wykonać następujący kod, który "Łączy" dwóch strumieni

• FIRST flatMapping do Stream<Stream<Integer>>
• następnie redukcję Stream<Stream<Integer>> pomocą Stream.concat()

I w obu przypadkach uzyskać taki sam poprawny wynik, ale liczba operacji filtrowania jest inna.

public class FlatMapVsReduce { 
    public static void main(String[] args) { 
     List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9); 

     Predicate<Integer> predicate1 = i -> { 
      System.out.println("testing first condition with " + i); 
      return i == 3; 
     }; 

     Predicate<Integer> predicate2 = i -> { 
      System.out.println("testing second condition with " + i); 
      return i == 7; 
     }; 

     System.out.println("Testing with flatMap"); 
     Integer result1 = 
      Stream.of(list.stream().filter(predicate1), 
         list.stream().filter(predicate2)) 
        .flatMap(Function.identity()) 
        .peek(i -> System.out.println("peeking " + i)) 
        .findFirst() 
        .orElse(null); 
     System.out.println("result1 = " + result1); 

     System.out.println(); 
     System.out.println("Testing with reduce"); 
     Integer result2 = 
      Stream.of(list.stream().filter(predicate1), 
         list.stream().filter(predicate2)) 
        .reduce(Stream::concat) 
        .orElseGet(Stream::empty) 
        .peek(i -> System.out.println("peeking " + i)) 
        .findFirst() 
        .orElse(null); 
     System.out.println("result2 = " + result2); 
    } 
} 

Otrzymuję oczekiwany wynik w obu przypadkach (3). Jednak pierwsza operacja dotyczy pierwszego filtru na każdym elemencie kolekcji, podczas gdy drugi zatrzymuje się, gdy tylko jeden zostanie spełniony. Dane wyjściowe to:

Testing with flatMap 
testing first condition with 1 
testing first condition with 2 
testing first condition with 3 
peeking 3 
testing first condition with 4 
testing first condition with 5 
testing first condition with 6 
testing first condition with 7 
testing first condition with 8 
testing first condition with 9 
result1 = 3 

Testing with reduce 
testing first condition with 1 
testing first condition with 2 
testing first condition with 3 
peeking 3 
result2 = 3 

Dlaczego różnica między nimi jest różna? Czy kod JDK mógłby zostać ulepszony, aby był tak wydajny w pierwszym scenariuszu, jak w drugim, czy jest coś w płaskiej mapie, które uniemożliwia?

Uzupełnienie: w następujący alternatywny jest tak wydajny jak ten przy użyciu zmniejszyć, ale wciąż nie potrafi wyjaśnić dlaczego:

Integer result3 = Stream.of(predicate1, predicate2) 
          .flatMap(c -> list.stream().filter(c).limit(1)) 
          .peek(i -> System.out.println("peeking " + i)) 
          .findFirst() 
          .orElse(null); 
    System.out.println("result3 = " + result3); 

Answer 1

z realizacji flatMap in openJDK, co rozumiem, jest to, że flatMap wypycha całą zawartość strumienia przychodzącego dołu:

result.sequential().forEach(downstreamAsInt); 

z drugiej strony, wydaje się być Stream::concat obsługi ściągania i wysyłania nie wszystko na raz.

Podejrzewam, że test nie pokazuje pełnego obrazu:

• W flatMap, drugi strumień jest rozważać jedynie wówczas gdy pierwszy jest wyczerpany.
• W reduce wszystkie strumienie są przesyłane w ostatnim połączonym strumieniu, ponieważ zredukowany obiekt nie ma sensu, dopóki cała zawartość strumienia wejściowego nie zostanie zużyta.

Co oznacza, że ​​użycie jednego z nich zależy od złożoności danych wejściowych. Jeśli masz nieskończone Stream<Stream<Integer>>, zmniejszenie nigdy się nie skończy.