Wie stoppen Streams?

Question

Ich frage mich, wenn ich mit meiner eigenen Stream.generate unendlichen Strom erzeugt, wie die Ströme, die in der Standardbibliothek Stopp sind ...

Zum Beispiel, wenn Sie eine Liste mit Datensätzen haben:

List<Record> records = getListWithRecords(); 
records.stream().forEach(/* do something */); 

Der Stream wird nicht unendlich sein und für immer laufen, aber es wird aufhören, wenn alle Elemente in der Liste durchlaufen werden. Aber wie funktioniert das? Die gleiche Funktionalität gilt für den von Files.lines(path) (Quelle: http://www.mkyong.com/java8/java-8-stream-read-a-file-line-by-line/) erstellt Strom.

Und eine zweite Frage, wie ein Strom mit Stream.generate erstellt wird, kann dann auf die gleiche Weise gestoppt werden?

Answer 1

Finite Ströme einfach nicht über Stream.generate erstellt.

Die Standardmethode zur Implementierung eines Streams ist die Implementierung eines Spliterator, manchmal unter Verwendung von the Iterator detour. In jedem Fall hat die Implementierung eine Möglichkeit, ein Ende zu melden, z. wenn Spliterator.tryAdvancefalse oder forEachRemaining zurückgibt, kehrt die Methode zurück, oder im Fall einer Iterator Quelle, wenn hasNext()false zurückgibt.

kann A Spliterator auch die erwartete Anzahl von Elementen berichten, bevor die Verarbeitung beginnt.

Bäche, über eine der Factory-Methoden innerhalb der Stream Schnittstelle geschaffen, wie Stream.generate kann entweder implementiert werden, von einem Spliterator auch oder mit internen Funktionen des Stroms Umsetzung, aber unabhängig davon, wie sie umgesetzt werden, Sie don‘ T Hände in dieser Implementierung, um ihr Verhalten zu ändern, also ist die einzige Möglichkeit, einen solchen Strom endlich zu machen, eine limit Operation an den Strom zu ketten.

Wenn Sie einen nicht leeren endlichen Stream erstellen möchten, der nicht von einem Array oder einer Sammlung unterstützt wird und keine der vorhandenen Streamquellen passt, müssen Sie Ihre eigenen Spliterator und create a stream out of it implementieren. Wie oben gesagt, können Sie eine vorhandene Methode verwenden, um ein Spliterator aus einem Iterator zu erstellen, aber Sie sollten die Versuchung widerstehen ein Iterator nur zu benutzen, weil es vertraut ist. Ein Spliterator ist nicht schwer zu implementieren:

/** like {@code Stream.generate}, but with an intrinsic limit */ 
static <T> Stream<T> generate(Supplier<T> s, long count) { 
    return StreamSupport.stream(
       new Spliterators.AbstractSpliterator<T>(count, Spliterator.SIZED) { 
     long remaining=count; 

     public boolean tryAdvance(Consumer<? super T> action) { 
      if(remaining<=0) return false; 
      remaining--; 
      action.accept(s.get()); 
      return true; 
     } 
    }, false); 
} 

Von diesem Ausgangspunkt können Sie Überschreibungen für die default Methoden des Spliterator Schnittstelle, Gewichtung Entwicklungsaufwand und mögliche Leistungsverbesserungen hinzuzufügen, z.B.

static <T> Stream<T> generate(Supplier<T> s, long count) { 
    return StreamSupport.stream(
       new Spliterators.AbstractSpliterator<T>(count, Spliterator.SIZED) { 
     long remaining=count; 

     public boolean tryAdvance(Consumer<? super T> action) { 
      if(remaining<=0) return false; 
      remaining--; 
      action.accept(s.get()); 
      return true; 
     } 

     /** May improve the performance of most non-short-circuiting operations */ 
     @Override 
     public void forEachRemaining(Consumer<? super T> action) { 
      long toGo=remaining; 
      remaining=0; 
      for(; toGo>0; toGo--) action.accept(s.get()); 
     } 
    }, false); 
} 

Answer 2

Ich habe eine generische Lösung für diesen erstellt

public class GuardedSpliterator<T> implements Spliterator<T> { 

    final Supplier<? extends T> generator; 

    final Predicate<T> termination; 

    final boolean inclusive; 

    public GuardedSpliterator(Supplier<? extends T> generator, Predicate<T> termination, boolean inclusive) { 
    this.generator = generator; 
    this.termination = termination; 
    this.inclusive = inclusive; 
    } 

    @Override 
    public boolean tryAdvance(Consumer<? super T> action) { 
    T next = generator.get(); 
    boolean end = termination.test(next); 
    if (inclusive || !end) { 
     action.accept(next); 
    } 
    return !end; 
    } 

    @Override 
    public Spliterator<T> trySplit() { 
    throw new UnsupportedOperationException("Not supported yet."); 
    } 

    @Override 
    public long estimateSize() { 
    throw new UnsupportedOperationException("Not supported yet."); 
    } 

    @Override 
    public int characteristics() { 
    return Spliterator.ORDERED; 
    } 

} 

Verbrauch ist recht einfach:

GuardedSpliterator<Integer> source = new GuardedSpliterator<>(
    () -> rnd.nextInt(), 
    (i) -> i > 10, 
    true 
); 

Stream<Integer> ints = StreamSupport.stream(source, false); 

ints.forEach(i -> System.out.println(i));