Limiter un flux par un prédicat

187

Existe-t-il une opération de flux Java 8 qui limite un (potentiellement infini) Stream jusqu'à ce que le premier élément ne corresponde pas à un prédicat?

En Java 9, nous pouvons utiliser takeWhilecomme dans l'exemple ci-dessous pour imprimer tous les nombres inférieurs à 10.

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

Comme il n'y a pas une telle opération dans Java 8, quelle est la meilleure façon de l'implémenter de manière générale?

MForster
la source
1
Informations éventuellement utiles sur: stackoverflow.com/q/19803058/248082
nobeh
Connexes: équivalent de Scala dropWhile
charlie
Je me demande comment les architectes pourraient jamais passer à travers le "à quoi pouvons-nous réellement faire usage " sans se heurter à ce cas d'utilisation. À partir de Java 8, les flux ne sont réellement utiles que pour les structures de données existantes: - /
Thorbjørn Ravn Andersen
Avec Java 9, il serait plus facile d'écrireIntStream.iterate(1, n->n<10, n->n+1).forEach(System.out::print);
Marc Dzaebel

Réponses:

81

Une telle opération devrait être possible avec un Java 8 Stream, mais elle ne peut pas nécessairement être effectuée efficacement - par exemple, vous ne pouvez pas nécessairement paralléliser une telle opération, car vous devez regarder les éléments dans l'ordre.

L'API ne fournit pas un moyen facile de le faire, mais le moyen le plus simple est probablement de prendre Stream.iterator(), d'encapsuler le Iteratorpour avoir une implémentation «à prendre en main», puis de revenir à a Spliterator, puis a Stream. Ou - peut-être - enveloppez le Spliterator, bien qu'il ne puisse plus vraiment être divisé dans cette implémentation.

Voici une implémentation non testée de takeWhilesur un 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
la source
8
En théorie, il est facile de paralléliser takeWhile avec un prédicat sans état. Évaluez la condition dans des lots parallèles (en supposant que le prédicat ne lance pas ou n'a pas d'effet secondaire s'il est exécuté plusieurs fois de plus). Le problème est de le faire dans le contexte de la décomposition récursive (framework fork / join) que Streams utilise. Vraiment, ce sont les Streams qui sont horriblement inefficaces.
Aleksandr Dubinsky
91
Les streams auraient été bien meilleurs s'ils n'avaient pas été aussi préoccupés par le parallélisme automagique. Le parallélisme n'est nécessaire que dans une infime partie des endroits où les flux peuvent être utilisés. De plus, si Oracle se souciait tellement de la performance, ils auraient pu faire l'autovectorisation de la JVM JIT et obtenir une amélioration beaucoup plus importante des performances, sans déranger les développeurs. Maintenant, le parallélisme automatique est bien fait.
Aleksandr Dubinsky
Vous devez mettre à jour cette réponse maintenant que Java 9 est disponible.
Radiodef
4
Non, @Radiodef. La question demande spécifiquement une solution Java 8.
Renato Back
146

Operations takeWhileet dropWhileont été ajoutés au JDK 9. Votre exemple de code

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

se comportera exactement comme vous vous y attendez lorsqu'il sera compilé et exécuté sous JDK 9.

Le JDK 9 a été publié. Il est disponible en téléchargement ici: http://jdk.java.net/9/

Marques Stuart
la source
3
Lien direct vers les documents de prévisualisation pour JDK9 Stream, avec takeWhile/ dropWhile: download.java.net/jdk9/docs/api/java/util/stream/Stream.html
Miles
1
Y a-t-il une raison pour laquelle ils sont appelés takeWhileet dropWhileplutôt que limitWhileet skipWhile, par souci de cohérence avec l'API existante?
Lukas Eder
10
@LukasEder takeWhileet dropWhilesont assez répandus, se produisant dans Scala, Python, Groovy, Ruby, Haskell et Clojure. L'asymétrie avec skipet limitest malheureuse. Peut skip- être et limitaurait dû être appelé dropet take, mais ceux-ci ne sont pas aussi intuitifs à moins que vous ne soyez déjà familier avec Haskell.
Stuart marque le
3
@StuartMarks: Je comprends cela dropXXXet ce takeXXXsont des termes plus populaires, mais je peux personnellement vivre avec les plus SQL-esque limitXXXet skipXXX. Je trouve cette nouvelle asymétrie beaucoup plus déroutante que le choix individuel des termes ... :) (btw: Scala a aussi drop(int)et take(int))
Lukas Eder
1
ouais laissez-moi juste passer à Jdk 9 en production. De nombreux développeurs sont toujours sur Jdk8, une telle fonctionnalité aurait dû être incluse avec Streams dès le début.
wilmol
50

allMatch()est une fonction de court-circuit, vous pouvez donc l'utiliser pour arrêter le traitement. Le principal inconvénient est que vous devez faire votre test deux fois: une fois pour voir si vous devez le traiter, et encore une fois pour voir s'il faut continuer.

IntStream
    .iterate(1, n -> n + 1)
    .peek(n->{if (n<10) System.out.println(n);})
    .allMatch(n->n < 10);
Michael Rowley
la source
5
Cela m'a semblé peu intuitif au début (étant donné le nom de la méthode), mais la documentation confirme qu'il Stream.allMatch()s'agit d'une opération de court-circuit . Donc, cela se terminera même sur un flux infini comme IntStream.iterate(). Bien sûr, rétrospectivement, il s'agit d'une optimisation sensée.
Bailey Parker
3
C'est bien, mais je ne pense pas qu'il communique très bien que son intention est le corps du peek. Si je le rencontrais le mois prochain, je prendrais une minute pour me demander pourquoi le programmeur avant moi a vérifié si allMatchpuis ignoré la réponse.
Joshua Goldberg
10
L'inconvénient de cette solution est qu'elle renvoie un booléen, vous ne pouvez donc pas collecter les résultats du flux comme vous le feriez normalement.
neXus
35

Dans le prolongement de la réponse @StuartMarks . Ma bibliothèque StreamEx a un takeWhilefonctionnement compatible avec l'implémentation actuelle du JDK-9. Lors de l'exécution sous JDK-9, il déléguera simplement à l'implémentation JDK (via MethodHandle.invokeExactlaquelle est vraiment rapide). Lors de l'exécution sous JDK-8, l'implémentation "polyfill" sera utilisée. Donc, en utilisant ma bibliothèque, le problème peut être résolu comme ceci:

IntStreamEx.iterate(1, n -> n + 1)
           .takeWhile(n -> n < 10)
           .forEach(System.out::println);
Tagir Valeev
la source
Pourquoi ne l'avez-vous pas implémenté pour la classe StreamEx?
Someguy
@Someguy je l'ai mis en œuvre.
Tagir Valeev
14

takeWhileest l'une des fonctions fournies par la bibliothèque protonpack .

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
la source
11

Mise à jour: Java 9 Streamest désormais livré avec une méthode takeWhile .

Pas besoin de hacks ou d'autres solutions. Utilisez juste ça!


Je suis sûr que cela peut être grandement amélioré: (quelqu'un pourrait peut-être le rendre thread-safe)

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

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

Un hack bien sûr ... Pas élégant - mais ça marche ~: 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);
    }

}
Le coordinateur
la source
8

Vous pouvez utiliser java8 + rxjava .

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));
fracasser
la source
6

En fait, il existe 2 façons de le faire dans Java 8 sans aucune bibliothèque supplémentaire ou en utilisant Java 9.

Si vous souhaitez imprimer les nombres de 2 à 20 sur la console, vous pouvez le faire:

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

ou

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

La sortie est dans les deux cas:

2
4
6
8
10
12
14
16
18
20

Personne n'a encore mentionné anyMatch . C'est la raison de cet article.

gil.fernandes
la source
5

Il s'agit de la source copiée à partir du JDK 9 java.util.stream.Stream.takeWhile (Predicate). Une petite différence pour travailler avec JDK 8.

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);
}
martien
la source
4

Voici une version réalisée sur ints - comme demandé dans la question.

Usage:

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

Voici le code pour 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
la source
2

Allez chercher la bibliothèque AbacusUtil . Il fournit l'API exacte que vous souhaitez et plus encore:

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

Déclaration : Je suis le développeur d'AbacusUtil.

user_3380739
la source
0

Vous ne pouvez pas abandonner un flux sauf par une opération de terminal en court-circuit, ce qui laisserait certaines valeurs de flux non traitées quelle que soit leur valeur. Mais si vous souhaitez simplement éviter les opérations sur un flux, vous pouvez ajouter une transformation et un filtre au flux:

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

Cela transforme le flux d'objets en valeurs nulles lorsque les éléments remplissent une condition, puis filtre les valeurs nulles. Si vous êtes prêt à vous adonner aux effets secondaires, vous pouvez définir la valeur de la condition sur true une fois que quelque chose est rencontré, de sorte que toutes les choses suivantes sont filtrées quelle que soit leur valeur. Mais même si ce n'est pas le cas, vous pouvez économiser beaucoup (sinon la totalité) du traitement en filtrant les valeurs du flux que vous ne souhaitez pas traiter.

Matthieu
la source
C'est nul qu'un évaluateur anonyme ait abaissé ma réponse sans dire pourquoi. Ainsi, ni moi ni aucun autre lecteur ne sait ce qui ne va pas avec ma réponse. En l'absence de leur justification, je considérerai leur critique invalide, et ma réponse comme publiée est correcte.
Matthew
Votre réponse ne résout pas le problème des OP, qui concerne des flux infinis. Cela semble également compliquer inutilement les choses car vous pouvez écrire la condition dans l'appel filter () lui-même, sans avoir besoin de map (). La question a déjà un exemple de code, essayez simplement d'appliquer votre réponse à ce code et vous verrez que le programme bouclera pour toujours.
SenoCtar
0

Même moi, j'avais une exigence similaire: invoquez le service Web, s'il échoue, réessayez 3 fois. S'il échoue même après ces nombreux essais, envoyez une notification par e-mail. Après avoir beaucoup cherché sur Google, anyMatch()est venu en sauveur. Mon exemple de code comme suit. Dans l'exemple suivant, si la méthode webServiceCall renvoie true dans la première itération elle-même, stream n'itère pas plus loin comme nous l'avons appelé anyMatch(). Je crois que c'est ce que vous recherchez.

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
la source
0

Si vous connaissez le nombre exact de répétitions qui seront effectuées, vous pouvez faire

IntStream
          .iterate(1, n -> n + 1)
          .limit(10)
          .forEach(System.out::println);
Dilip Tharoor
la source
1
Bien que cela puisse répondre à la question des auteurs, il manque des mots explicatifs et des liens vers la documentation. Les extraits de code bruts ne sont pas très utiles sans quelques phrases autour. Vous pouvez également trouver comment rédiger une bonne réponse très utile. Veuillez modifier votre réponse.
hellow
0
    IntStream.iterate(1, n -> n + 1)
    .peek(System.out::println) //it will be executed 9 times
    .filter(n->n>=9)
    .findAny();

au lieu de peak, vous pouvez utiliser mapToObj pour renvoyer l'objet ou le message final

    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 Potomkine
la source
-2

Si vous avez un problème différent, une solution différente peut être nécessaire, mais pour votre problème actuel, j'irais simplement avec:

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

Peut-être un peu hors sujet, mais c'est ce que nous avons pour List<T>plutôt queStream<T> .

Vous devez d'abord avoir une takeméthode util. Cette méthode prend les premiers néléments:

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

ça marche juste comme 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));

maintenant, il sera assez simple d'écrire une takeWhileméthode basée surtake

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
}

ça marche comme ça:

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

cette implémentation itère partiellement la liste plusieurs fois mais n'ajoute pas d' O(n^2)opérations d' ajout . J'espère que c'est acceptable.

Max
la source
-3

J'ai une autre solution rapide en mettant en œuvre ceci (ce qui est plutôt impur en fait, mais vous voyez l'idée):

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
la source
2
Vous évaluez l' ensemble flux à l'avance! Et si currentjamais .equals(e), vous obtiendrez une boucle sans fin. Les deux même si vous postulez par exemple .limit(1). C'est bien pire que «impur» .
charlie
-3

Voici ma tentative en utilisant uniquement la bibliothèque Java Stream.

        IntStream.iterate(0, i -> i + 1)
        .filter(n -> {
                if (n < 10) {
                    System.out.println(n);
                    return false;
                } else {
                    return true;
                }
            })
        .findAny();
album_escalade
la source
Le filterprédicat est censé être apatride. System.out.printlnest un effet secondaire.
Radiodef