Begrenzen Sie einen Stream durch ein Prädikat

187

Gibt es eine Java 8-Stream-Operation, die a (möglicherweise unendlich) begrenzt, Streambis das erste Element nicht mehr mit einem Prädikat übereinstimmt?

In Java 9 können wir takeWhilewie im folgenden Beispiel alle Zahlen unter 10 drucken.

IntStream
    .iterate(1, n -> n + 1)
    .takeWhile(n -> n < 10)
    .forEach(System.out::println);

Da es in Java 8 keine solche Operation gibt, wie kann sie allgemein am besten implementiert werden?

MForster
quelle
1
Möglicherweise nützliche Informationen unter: stackoverflow.com/q/19803058/248082
nobeh
Verwandte: Äquivalent von Scala dropWhile
Charlie
Ich frage mich, wie die Architekten jemals durch das "Wofür können wir das tatsächlich verwenden " kommen könnten, ohne auf diesen Anwendungsfall zu stoßen. Ab Java 8 sind Streams eigentlich nur für bestehende Datenstrukturen hilfreich: - /
Thorbjørn Ravn Andersen
Mit Java 9 wäre es einfacher zu schreibenIntStream.iterate(1, n->n<10, n->n+1).forEach(System.out::print);
Marc Dzaebel

Antworten:

81

Eine solche Operation sein sollte möglich mit einem Java - 8 Stream, aber es kann nicht notwendigerweise effizient durchgeführt werden - zum Beispiel, kann man nicht unbedingt parallelisieren eine solche Operation, wie Sie auf Elemente, um zu suchen.

Die API bietet keine einfache Möglichkeit, dies zu tun, aber die wahrscheinlich einfachste Möglichkeit besteht darin Stream.iterator(), die IteratorImplementierung für eine "Take-while" -Implementierung zu verpacken und dann zu a Spliteratorund dann zu a zurückzukehren Stream. Oder - vielleicht - wickeln Sie das ein Spliterator, obwohl es in dieser Implementierung nicht mehr wirklich aufgeteilt werden kann.

Hier ist eine ungetestete Implementierung von takeWhile a Spliterator:

static <T> Spliterator<T> takeWhile(
    Spliterator<T> splitr, Predicate<? super T> predicate) {
  return new Spliterators.AbstractSpliterator<T>(splitr.estimateSize(), 0) {
    boolean stillGoing = true;
    @Override public boolean tryAdvance(Consumer<? super T> consumer) {
      if (stillGoing) {
        boolean hadNext = splitr.tryAdvance(elem -> {
          if (predicate.test(elem)) {
            consumer.accept(elem);
          } else {
            stillGoing = false;
          }
        });
        return hadNext && stillGoing;
      }
      return false;
    }
  };
}

static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> predicate) {
   return StreamSupport.stream(takeWhile(stream.spliterator(), predicate), false);
}
Louis Wasserman
quelle
8
Theoretisch ist es einfach, takeWhile mit einem zustandslosen Prädikat zu parallelisieren. Bewerten Sie den Zustand in parallelen Stapeln (vorausgesetzt, das Prädikat löst nicht aus oder hat keine Nebenwirkungen, wenn es einige zusätzliche Male ausgeführt wird). Das Problem besteht darin, dass es im Kontext der rekursiven Zerlegung (Fork / Join-Framework) verwendet wird, die Streams verwenden. Wirklich, es sind Streams, die schrecklich ineffizient sind.
Aleksandr Dubinsky
91
Streams wären viel besser gewesen, wenn sie sich nicht so sehr mit automatischer Parallelität beschäftigt hätten. Parallelität ist nur an einem winzigen Bruchteil der Stellen erforderlich, an denen Streams verwendet werden können. Wenn Oracle sich so sehr um Leistung gekümmert hätte, hätten sie außerdem die JVM-JIT-Autovektorisierung durchführen und einen viel größeren Leistungsschub erzielen können, ohne die Entwickler zu stören. Nun, das ist automagische Parallelität, die richtig gemacht wird.
Aleksandr Dubinsky
Sie sollten diese Antwort jetzt aktualisieren, nachdem Java 9 veröffentlicht wurde.
Radiodef
4
Nein, @Radiodef. Die Frage fragt speziell nach einer Java 8-Lösung.
Renato Zurück
145

Operationen takeWhileund dropWhilewurden zu JDK 9 hinzugefügt. Ihr Beispielcode

IntStream
    .iterate(1, n -> n + 1)
    .takeWhile(n -> n < 10)
    .forEach(System.out::println);

verhält sich beim Kompilieren und Ausführen unter JDK 9 genau so, wie Sie es erwarten.

JDK 9 wurde veröffentlicht. Es kann hier heruntergeladen werden: http://jdk.java.net/9/

Stuart Marks
quelle
3
Direkter Link zu den Vorschaudokumenten für JDK9 Stream mit takeWhile/ dropWhile: download.java.net/jdk9/docs/api/java/util/stream/Stream.html
Meilen
1
Gibt es einen Grund, warum sie aufgerufen werden takeWhileund dropWhilenicht limitWhileund skipWhile, um die Konsistenz mit der vorhandenen API zu gewährleisten?
Lukas Eder
10
@LukasEder takeWhileund dropWhilesind ziemlich weit verbreitet und kommen in Scala, Python, Groovy, Ruby, Haskell und Clojure vor. Die Asymmetrie mit skipund limitist unglücklich. Vielleicht skipund limitsollte genannt worden dropund take, aber die sind nicht so intuitiv , wenn Sie mit Haskell bereits vertraut sind.
Stuart Marks
3
@StuartMarks: Ich verstehe das dropXXXund takeXXXsind populärere Begriffe, aber ich kann persönlich mit den SQL-ähnlichen limitXXXund leben skipXXX. Ich finde diese neue Asymmetrie viel verwirrender als die individuelle Wahl der Begriffe ... :) (übrigens: Scala hat auch drop(int)und take(int))
Lukas Eder
1
Ja, lassen Sie mich einfach in der Produktion auf Jdk 9 upgraden. Viele Entwickler sind noch auf Jdk8, eine solche Funktion sollte von Anfang an in Streams enthalten sein.
Wilmol
50

allMatch()ist eine Kurzschlussfunktion, mit der Sie die Verarbeitung stoppen können. Der Hauptnachteil ist, dass Sie Ihren Test zweimal durchführen müssen: einmal, um zu sehen, ob Sie ihn verarbeiten sollten, und erneut, um zu sehen, ob Sie weitermachen sollen.

IntStream
    .iterate(1, n -> n + 1)
    .peek(n->{if (n<10) System.out.println(n);})
    .allMatch(n->n < 10);
Michael Rowley
quelle
5
Dies schien mir zunächst (angesichts des Methodennamens) nicht intuitiv zu sein, aber die Dokumente bestätigen, dass Stream.allMatch()es sich um einen Kurzschluss handelt . Dies wird also auch bei einem unendlichen Stream wie abgeschlossen sein IntStream.iterate(). Rückblickend ist dies natürlich eine sinnvolle Optimierung.
Bailey Parker
3
Das ist ordentlich, aber ich denke nicht, dass es sehr gut kommuniziert, dass seine Absicht der Körper des ist peek. Wenn ich nächsten Monat darauf stoßen würde, würde ich mir eine Minute Zeit nehmen, um mich zu fragen, warum der Programmierer vor mir überprüft hat, ob allMatchund dann die Antwort ignoriert hat.
Joshua Goldberg
10
Der Nachteil dieser Lösung besteht darin, dass sie einen Booleschen Wert zurückgibt, sodass Sie die Ergebnisse des Streams nicht wie gewohnt erfassen können.
neXus
35

Als Folge der Antwort von @StuartMarks . Meine StreamEx- Bibliothek verfügt über den takeWhileVorgang, der mit der aktuellen JDK-9-Implementierung kompatibel ist. Wenn es unter JDK-9 ausgeführt wird, wird es nur an die JDK-Implementierung delegiert (über MethodHandle.invokeExactdie es sehr schnell geht). Bei Ausführung unter JDK-8 wird die Implementierung "Polyfill" verwendet. Mit meiner Bibliothek kann das Problem also folgendermaßen gelöst werden:

IntStreamEx.iterate(1, n -> n + 1)
           .takeWhile(n -> n < 10)
           .forEach(System.out::println);
Tagir Valeev
quelle
Warum haben Sie es nicht für die StreamEx-Klasse implementiert?
Someguy
@ Someguy Ich habe es implementiert.
Tagir Valeev
14

takeWhileist eine der Funktionen der Protonpack-Bibliothek .

Stream<Integer> infiniteInts = Stream.iterate(0, i -> i + 1);
Stream<Integer> finiteInts = StreamUtils.takeWhile(infiniteInts, i -> i < 10);

assertThat(finiteInts.collect(Collectors.toList()),
           hasSize(10));
Dominic Fox
quelle
11

Update: Java 9 enthält Streamjetzt eine takeWhile- Methode.

Keine Notwendigkeit für Hacks oder andere Lösungen. Benutze das einfach!


Ich bin sicher, dass dies erheblich verbessert werden kann: (Jemand könnte es vielleicht threadsicher machen)

Stream<Integer> stream = Stream.iterate(0, n -> n + 1);

TakeWhile.stream(stream, n -> n < 10000)
         .forEach(n -> System.out.print((n == 0 ? "" + n : "," + n)));

Ein Hack sicher ... Nicht elegant - aber es funktioniert ~: D.

class TakeWhile<T> implements Iterator<T> {

    private final Iterator<T> iterator;
    private final Predicate<T> predicate;
    private volatile T next;
    private volatile boolean keepGoing = true;

    public TakeWhile(Stream<T> s, Predicate<T> p) {
        this.iterator = s.iterator();
        this.predicate = p;
    }

    @Override
    public boolean hasNext() {
        if (!keepGoing) {
            return false;
        }
        if (next != null) {
            return true;
        }
        if (iterator.hasNext()) {
            next = iterator.next();
            keepGoing = predicate.test(next);
            if (!keepGoing) {
                next = null;
            }
        }
        return next != null;
    }

    @Override
    public T next() {
        if (next == null) {
            if (!hasNext()) {
                throw new NoSuchElementException("Sorry. Nothing for you.");
            }
        }
        T temp = next;
        next = null;
        return temp;
    }

    public static <T> Stream<T> stream(Stream<T> s, Predicate<T> p) {
        TakeWhile tw = new TakeWhile(s, p);
        Spliterator split = Spliterators.spliterator(tw, Integer.MAX_VALUE, Spliterator.ORDERED);
        return StreamSupport.stream(split, false);
    }

}
Der Koordinator
quelle
8

Sie können java8 + rxjava verwenden .

import java.util.stream.IntStream;
import rx.Observable;


// Example 1)
IntStream intStream  = IntStream.iterate(1, n -> n + 1);
Observable.from(() -> intStream.iterator())
    .takeWhile(n ->
          {
                System.out.println(n);
                return n < 10;
          }
    ).subscribe() ;


// Example 2
IntStream intStream  = IntStream.iterate(1, n -> n + 1);
Observable.from(() -> intStream.iterator())
    .takeWhile(n -> n < 10)
    .forEach( n -> System.out.println(n));
frhack
quelle
6

Tatsächlich gibt es in Java 8 zwei Möglichkeiten, dies ohne zusätzliche Bibliotheken oder mit Java 9 zu tun.

Wenn Sie Zahlen von 2 bis 20 auf der Konsole drucken möchten, können Sie dies tun:

IntStream.iterate(2, (i) -> i + 2).peek(System.out::println).allMatch(i -> i < 20);

oder

IntStream.iterate(2, (i) -> i + 2).peek(System.out::println).anyMatch(i -> i >= 20);

Die Ausgabe erfolgt in beiden Fällen:

2
4
6
8
10
12
14
16
18
20

Noch hat niemand anyMatch erwähnt . Dies ist der Grund für diesen Beitrag.

gil.fernandes
quelle
5

Dies ist die aus JDK 9 kopierte Quelle java.util.stream.Stream.takeWhile (Predicate). Ein kleiner Unterschied, um mit JDK 8 zu arbeiten.

static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> p) {
    class Taking extends Spliterators.AbstractSpliterator<T> implements Consumer<T> {
        private static final int CANCEL_CHECK_COUNT = 63;
        private final Spliterator<T> s;
        private int count;
        private T t;
        private final AtomicBoolean cancel = new AtomicBoolean();
        private boolean takeOrDrop = true;

        Taking(Spliterator<T> s) {
            super(s.estimateSize(), s.characteristics() & ~(Spliterator.SIZED | Spliterator.SUBSIZED));
            this.s = s;
        }

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            boolean test = true;
            if (takeOrDrop &&               // If can take
                    (count != 0 || !cancel.get()) && // and if not cancelled
                    s.tryAdvance(this) &&   // and if advanced one element
                    (test = p.test(t))) {   // and test on element passes
                action.accept(t);           // then accept element
                return true;
            } else {
                // Taking is finished
                takeOrDrop = false;
                // Cancel all further traversal and splitting operations
                // only if test of element failed (short-circuited)
                if (!test)
                    cancel.set(true);
                return false;
            }
        }

        @Override
        public Comparator<? super T> getComparator() {
            return s.getComparator();
        }

        @Override
        public void accept(T t) {
            count = (count + 1) & CANCEL_CHECK_COUNT;
            this.t = t;
        }

        @Override
        public Spliterator<T> trySplit() {
            return null;
        }
    }
    return StreamSupport.stream(new Taking(stream.spliterator()), stream.isParallel()).onClose(stream::close);
}
Marsmensch
quelle
4

Hier ist eine Version von Ints - wie in der Frage gestellt.

Verwendung:

StreamUtil.takeWhile(IntStream.iterate(1, n -> n + 1), n -> n < 10);

Hier ist der Code für StreamUtil:

import java.util.PrimitiveIterator;
import java.util.Spliterators;
import java.util.function.IntConsumer;
import java.util.function.IntPredicate;
import java.util.stream.IntStream;
import java.util.stream.StreamSupport;

public class StreamUtil
{
    public static IntStream takeWhile(IntStream stream, IntPredicate predicate)
    {
        return StreamSupport.intStream(new PredicateIntSpliterator(stream, predicate), false);
    }

    private static class PredicateIntSpliterator extends Spliterators.AbstractIntSpliterator
    {
        private final PrimitiveIterator.OfInt iterator;
        private final IntPredicate predicate;

        public PredicateIntSpliterator(IntStream stream, IntPredicate predicate)
        {
            super(Long.MAX_VALUE, IMMUTABLE);
            this.iterator = stream.iterator();
            this.predicate = predicate;
        }

        @Override
        public boolean tryAdvance(IntConsumer action)
        {
            if (iterator.hasNext()) {
                int value = iterator.nextInt();
                if (predicate.test(value)) {
                    action.accept(value);
                    return true;
                }
            }

            return false;
        }
    }
}
Chris Greenaway
quelle
2

Gehen Sie zur Bibliothek AbacusUtil . Es bietet genau die gewünschte API und mehr:

IntStream.iterate(1, n -> n + 1).takeWhile(n -> n < 10).forEach(System.out::println);

Erklärung: Ich bin der Entwickler von AbacusUtil.

user_3380739
quelle
0

Sie können einen Stream nur durch einen Kurzschluss des Terminals abbrechen, wodurch einige Stream-Werte unabhängig von ihrem Wert unverarbeitet bleiben. Wenn Sie jedoch nur Operationen an einem Stream vermeiden möchten, können Sie dem Stream eine Transformation und einen Filter hinzufügen:

import java.util.Objects;

class ThingProcessor
{
    static Thing returnNullOnCondition(Thing thing)
    {    return( (*** is condition met ***)? null : thing);    }

    void processThings(Collection<Thing> thingsCollection)
    {
        thingsCollection.stream()
        *** regular stream processing ***
        .map(ThingProcessor::returnNullOnCondition)
        .filter(Objects::nonNull)
        *** continue stream processing ***
    }
} // class ThingProcessor

Das wandelt den Strom von Dingen in Nullen um, wenn die Dinge eine Bedingung erfüllen, und filtert dann Nullen heraus. Wenn Sie bereit sind, sich Nebenwirkungen hinzugeben, können Sie den Bedingungswert auf true setzen, sobald etwas auftritt, sodass alle nachfolgenden Dinge unabhängig von ihrem Wert herausgefiltert werden. Aber selbst wenn nicht, können Sie viel (wenn nicht sogar die gesamte) Verarbeitung sparen, indem Sie Werte aus dem Stream herausfiltern, die Sie nicht verarbeiten möchten.

Matthew
quelle
Es ist lahm, dass ein anonymer Bewerter meine Antwort herabgestuft hat, ohne zu sagen warum. Also wissen weder ich noch irgendein anderer Leser, was mit meiner Antwort falsch ist. In Ermangelung ihrer Rechtfertigung halte ich ihre Kritik für ungültig und meine Antwort als richtig.
Matthew
Ihre Antwort löst nicht das OP-Problem, das sich mit unendlichen Streams befasst. Es scheint auch die Dinge unnötig zu komplizieren, da Sie die Bedingung in den filter () -Aufruf selbst schreiben können, ohne map () zu benötigen. Die Frage enthält bereits einen Beispielcode. Versuchen Sie einfach, Ihre Antwort auf diesen Code anzuwenden, und Sie werden sehen, dass das Programm für immer wiederholt wird.
SenoCtar
0

Sogar ich hatte eine ähnliche Anforderung: Rufen Sie den Webdienst auf. Wenn dies fehlschlägt, wiederholen Sie ihn dreimal. Wenn dies auch nach diesen vielen Versuchen fehlschlägt, senden Sie eine E-Mail-Benachrichtigung. Nachdem ich viel gegoogelt hatte, anyMatch()kam ich als Retter. Mein Beispielcode wie folgt. Wenn im folgenden Beispiel die webServiceCall- Methode in der ersten Iteration selbst true zurückgibt, iteriert der Stream nicht weiter, wie wir es aufgerufen haben anyMatch(). Ich glaube, das ist es, wonach Sie suchen.

import java.util.stream.IntStream;

import io.netty.util.internal.ThreadLocalRandom;

class TrialStreamMatch {

public static void main(String[] args) {        
    if(!IntStream.range(1,3).anyMatch(integ -> webServiceCall(integ))){
         //Code for sending email notifications
    }
}

public static boolean webServiceCall(int i){
    //For time being, I have written a code for generating boolean randomly
    //This whole piece needs to be replaced by actual web-service client code
    boolean bool = ThreadLocalRandom.current().nextBoolean();
    System.out.println("Iteration index :: "+i+" bool :: "+bool);

    //Return success status -- true or false
    return bool;
}
Chinmay Phadke
quelle
0

Wenn Sie genau wissen, wie viele Wiederholungen durchgeführt werden, können Sie dies tun

IntStream
          .iterate(1, n -> n + 1)
          .limit(10)
          .forEach(System.out::println);
Dilip Tharoor
quelle
1
Dies könnte zwar die Frage des Autors beantworten, es fehlen jedoch einige erklärende Wörter und Links zur Dokumentation. Rohcode-Schnipsel sind ohne einige Ausdrücke nicht sehr hilfreich. Möglicherweise ist es auch sehr hilfreich , eine gute Antwort zu schreiben . Bitte bearbeiten Sie Ihre Antwort.
Gelb
0
    IntStream.iterate(1, n -> n + 1)
    .peek(System.out::println) //it will be executed 9 times
    .filter(n->n>=9)
    .findAny();

Anstelle von Peak können Sie mapToObj verwenden, um das endgültige Objekt oder die endgültige Nachricht zurückzugeben

    IntStream.iterate(1, n -> n + 1)
    .mapToObj(n->{   //it will be executed 9 times
            if(n<9)
                return "";
            return "Loop repeats " + n + " times";});
    .filter(message->!message.isEmpty())
    .findAny()
    .ifPresent(System.out::println);
Oleksandr Potomkin
quelle
-2

Wenn Sie ein anderes Problem haben, ist möglicherweise eine andere Lösung erforderlich, aber für Ihr aktuelles Problem würde ich einfach gehen mit:

IntStream
    .iterate(1, n -> n + 1)
    .limit(10)
    .forEach(System.out::println);
krmanish007
quelle
-2

Könnte ein bisschen vom Thema abweichen, aber das ist es, wofür wir es haben List<T>und nichtStream<T> .

Zuerst benötigen Sie eine takeutil-Methode. Diese Methode verwendet erste nElemente:

static <T> List<T> take(List<T> l, int n) {
    if (n <= 0) {
        return newArrayList();
    } else {
        int takeTo = Math.min(Math.max(n, 0), l.size());
        return l.subList(0, takeTo);
    }
}

es funktioniert einfach so scala.List.take

    assertEquals(newArrayList(1, 2, 3), take(newArrayList(1, 2, 3, 4, 5), 3));
    assertEquals(newArrayList(1, 2, 3), take(newArrayList(1, 2, 3), 5));

    assertEquals(newArrayList(), take(newArrayList(1, 2, 3), -1));
    assertEquals(newArrayList(), take(newArrayList(1, 2, 3), 0));

Jetzt wird es ziemlich einfach sein, eine takeWhileMethode zu schreiben , die auf basierttake

static <T> List<T> takeWhile(List<T> l, Predicate<T> p) {
    return l.stream().
            filter(p.negate()).findFirst(). // find first element when p is false
            map(l::indexOf).        // find the index of that element
            map(i -> take(l, i)).   // take up to the index
            orElse(l);  // return full list if p is true for all elements
}

es funktioniert so:

    assertEquals(newArrayList(1, 2, 3), takeWhile(newArrayList(1, 2, 3, 4, 3, 2, 1), i -> i < 4));

Diese Implementierung iteriert die Liste einige Male teilweise, fügt jedoch keine zusätzlichen O(n^2)Operationen hinzu. Hoffe das ist akzeptabel.

Max
quelle
-3

Ich habe eine andere schnelle Lösung, indem ich dies implementiere (was in der Tat ziemlich unrein ist, aber Sie bekommen die Idee):

public static void main(String[] args) {
    System.out.println(StreamUtil.iterate(1, o -> o + 1).terminateOn(15)
            .map(o -> o.toString()).collect(Collectors.joining(", ")));
}

static interface TerminatedStream<T> {
    Stream<T> terminateOn(T e);
}

static class StreamUtil {
    static <T> TerminatedStream<T> iterate(T seed, UnaryOperator<T> op) {
        return new TerminatedStream<T>() {
            public Stream<T> terminateOn(T e) {
                Builder<T> builder = Stream.<T> builder().add(seed);
                T current = seed;
                while (!current.equals(e)) {
                    current = op.apply(current);
                    builder.add(current);
                }
                return builder.build();
            }
        };
    }
}
user2504380
quelle
2
Sie bewerten den gesamten Stream im Voraus! Und wenn currentnie .equals(e), erhalten Sie eine Endlosschleife. Beides auch wenn Sie sich später bewerben zB .limit(1). Das ist weitaus schlimmer als "unrein" .
Charlie
-3

Hier ist mein Versuch, nur die Java Stream-Bibliothek zu verwenden.

        IntStream.iterate(0, i -> i + 1)
        .filter(n -> {
                if (n < 10) {
                    System.out.println(n);
                    return false;
                } else {
                    return true;
                }
            })
        .findAny();
Klettern_bum
quelle
Das filterPrädikat soll staatenlos sein. System.out.printlnist eine Nebenwirkung.
Radiodef