Flusso <Stream>: flatMap vs ridurre

Question

Se eseguo il seguente codice che "concatena" due flussi

• decrescente per flatMapping un Stream<Stream<Integer>>
• poi riducendo un Stream<Stream<Integer>> utilizzando Stream.concat()

I ottenere lo stesso risultato corretto in entrambi i casi, ma il numero di operazioni di filtro è diverso.

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

Ottengo il risultato previsto in entrambi i casi (3). Tuttavia, la prima operazione applica il primo filtro su ogni elemento della raccolta, mentre il secondo si interrompe non appena viene raggiunto uno. L'output è:

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 

Perché c'è una differenza di comportamento tra i due? Il codice JDK potrebbe essere migliorato per essere efficiente nel primo scenario rispetto al secondo, oppure c'è qualcosa in flatMap che lo rende impossibile?

Addendum: la seguente alternativa è efficiente come quella che utilizza ridurre, ma ancora non può spiegare perché:

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

Dalla realizzazione di flatMap in openJDK, quello che ho capito è che flatMap spinge l'intero contenuto del flusso in ingresso a valle:

result.sequential().forEach(downstreamAsInt); 

D'altra parte, Stream::concat sembra essere la manipolazione tirare e non invio tutto in una volta.

Ho il sospetto che il test non mostra il quadro completo:

• Nel flatMap, il secondo flusso solo è considerato quando la prima è esaurita.
• In reduce, tutti gli stream vengono spinti nello stream concatenato finale, perché l'oggetto ridotto non ha senso finché non viene consumato tutto il contenuto del flusso di input.

Il che significa che l'utilizzo di uno o dell'altro dipende dalla complessità degli input. Se hai un infinito Stream<Stream<Integer>>, la riduzione non finirà mai.