Für Stream.generate (foo) .parallel(), muss foo threadsicher sein?

Question

Ich arbeite mit Java 8 Streams. Ich habe eine benutzerdefinierte Funktion, foo(), die ein Objekt ergibt, und ich möchte die Objekte, die es erstellt, parallel streamen. Ich weiß, foo() ist nicht Thread sicher. Wenn ich Stream.generate(foo).parallel() schreibe, wird foo() asynchron aufgerufen? I.e. Werden Objekte seriell erzeugt und an parallele Threads übergeben, oder werden mehrere Threads je nach Bedarf Objekte erzeugen, indem sie foo()?

Answer 1

Während Daten Rennen kein garantiertes Verhalten sind, der folgende Code

System.out.println(
    Stream.generate(new Supplier<Integer>() { 
     int i; @Override public Integer get() { return i++; } 
    }).parallel() 
     .limit(10_000) 
     .collect(BitSet::new, BitSet::set, BitSet::or) 
     .cardinality() 
); 

Zahlen reproduzierbar druckt weniger als 10.000 in meiner Umgebung, was zeigt, dass fehlende Updates tatsächlich passieren können, wenn der Lieferant nicht Thread-sicher ist.

Beachten Sie, dass es auch möglich ist, dass der Lieferant nach mehr Elementen gefragt wird, als für die Ergebnisauswertung benötigt werden. Z.B.

LongAdder adder = new LongAdder(); 
System.out.println(
    Stream.generate(new Supplier<Integer>() { 
     int i; @Override public Integer get() { adder.increment(); return i++; } 
    }).parallel() 
     .limit(10_000) 
     .collect(BitSet::new, BitSet::set, BitSet::or) 
     .cardinality() 
); 
System.out.println("queried "+adder+" times"); 

in der Regel berichtet über eine Anzahl von Anfragen von mehr als 10.000, während zugleich die Ergebnisberichte weniger als 10.000 verschiedene Elemente aufgrund der Daten Rennen.

Wenn Sie den Lieferanten-Thread sicher machen, wird das Ergebnis auf die korrekte Anzahl von 10.000 verschiedenen Elementen geändert, aber der Lieferant kann immer noch mehr als 10.000 Mal abgefragt werden. Daher kann das Ergebnis nicht genau die Zahlen von 0 bis 9.999 enthalten , da der über generate erzeugte Strom ungeordnet ist, so können alle 10.000 verschiedenen Nummern vom Lieferanten verwendet werden.

Answer 2

Der Lieferant wird von mehrer Threads aufgerufen werden, wie Sie mit einem schnellen Experiment beobachten können:

Stream.generate(() -> Thread.currentThread().getId()) 
    .parallel() 
    .limit(100000) 
    .distinct() 
    .forEach(System.out::println);