Limitar una secuencia por un predicado

187

¿Existe una operación de flujo Java 8 que limite un (potencialmente infinito) Streamhasta que el primer elemento no coincida con un predicado?

En Java 9 podemos usar takeWhilecomo en el ejemplo a continuación para imprimir todos los números menores que 10.

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

Como no existe tal operación en Java 8, ¿cuál es la mejor manera de implementarlo de manera general?

MForster
fuente
1
Información posiblemente útil en: stackoverflow.com/q/19803058/248082
nobeh
Relacionado: Equivalente de Scala dropWhile
charlie
Me pregunto cómo los arquitectos podrían superar el "para qué podemos usar esto realmente " sin toparse con este caso de uso. A partir de Java 8, las secuencias solo son realmente útiles para las estructuras de datos existentes: - /
Thorbjørn Ravn Andersen
Con Java 9, sería más fácil escribirIntStream.iterate(1, n->n<10, n->n+1).forEach(System.out::print);
Marc Dzaebel

Respuestas:

81

Tal operación debería ser posible con un Java 8 Stream, pero no necesariamente se puede hacer de manera eficiente; por ejemplo, no necesariamente se puede paralelizar dicha operación, ya que hay que mirar los elementos en orden.

La API no proporciona una manera fácil de hacerlo, pero lo que probablemente sea la forma más sencilla es tomar Stream.iterator(), envolver Iteratorpara tener una implementación de "toma de tiempo", y luego volver a ay Spliteratorluego a Stream. O, tal vez, envolverlo Spliterator, aunque en realidad ya no se puede dividir en esta implementación.

Aquí hay una implementación no probada de takeWhileen 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
fuente
8
En teoría, la paralelización de takeWhile con un predicado sin estado es fácil. Evalúe la condición en lotes paralelos (suponiendo que el predicado no arroje o tenga un efecto secundario si se ejecuta algunas veces más). El problema es hacerlo en el contexto de la descomposición recursiva (fork / join framework) que utilizan Streams. Realmente, son las secuencias que son terriblemente ineficientes.
Aleksandr Dubinsky
9191
Las transmisiones habrían sido mucho mejores si no estuvieran tan preocupados por el paralelismo automágico. El paralelismo se necesita solo en una pequeña fracción de los lugares donde se pueden usar Streams. Además, si Oracle se preocupara tanto por la perforación, podrían haber hecho que el JVM JIT se autovectorizara y obtuvieran un aumento de rendimiento mucho mayor, sin molestar a los desarrolladores. Ahora que es el paralelismo automágico bien hecho.
Aleksandr Dubinsky
Debería actualizar esta respuesta ahora que se lanzó Java 9.
Radiodef
44
No, @Radiodef. La pregunta pide específicamente una solución Java 8.
Renato Volver
146

Operaciones takeWhiley dropWhilese han agregado a JDK 9. Su código de ejemplo

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

se comportará exactamente como lo espera cuando se compila y ejecuta bajo JDK 9.

JDK 9 ha sido lanzado. Está disponible para descargar aquí: http://jdk.java.net/9/

Stuart Marks
fuente
3
Enlace directo a los documentos de vista previa para JDK9 Stream, con takeWhile/ dropWhile: download.java.net/jdk9/docs/api/java/util/stream/Stream.html
Millas del
1
¿Hay alguna razón por la que se les llama takeWhiley en dropWhilelugar de limitWhiley skipWhile, por coherencia con la API existente?
Lukas Eder
10
@LukasEder takeWhiley dropWhileestán bastante extendidos, ocurren en Scala, Python, Groovy, Ruby, Haskell y Clojure. La asimetría con skipy limites lamentable. Tal skipy limitdebería haber sido llamado dropy take, pero esos no son tan intuitivos a menos que ya esté familiarizado con Haskell.
Stuart Marks
3
@StuartMarks: Entiendo eso dropXXXy takeXXXson términos más populares, pero personalmente puedo vivir con más SQL-esque limitXXXy skipXXX. Encuentro esta nueva asimetría mucho más confusa que la elección individual de términos ... :) (por cierto: Scala también tiene drop(int)y take(int))
Lukas Eder
1
sí, permítanme actualizarme a Jdk 9 en producción. Muchos desarrolladores todavía están en Jdk8, esa característica debería haberse incluido con Streams desde el principio.
wilmol
50

allMatch()es una función de cortocircuito, por lo que puede usarla para detener el procesamiento. La principal desventaja es que debe hacer su prueba dos veces: una para ver si debe procesarla y otra vez para ver si debe continuar.

IntStream
    .iterate(1, n -> n + 1)
    .peek(n->{if (n<10) System.out.println(n);})
    .allMatch(n->n < 10);
Michael Rowley
fuente
55
Al principio, esto no me pareció intuitivo (dado el nombre del método), pero los documentos confirman que Stream.allMatch()es una operación de cortocircuito . Entonces esto se completará incluso en una secuencia infinita como IntStream.iterate(). Por supuesto, en retrospectiva, esta es una optimización sensata.
Bailey Parker
3
Esto es bueno, pero no creo que comunique muy bien que su intención es el cuerpo del peek. Si lo encontraba el próximo mes, me tomaría un minuto preguntarme por qué el programador que tenía delante verificó si allMatchignoraba la respuesta.
Joshua Goldberg
10
La desventaja de esta solución es que devuelve un valor booleano, por lo que no puede recopilar los resultados de la secuencia como lo haría normalmente.
neXus
35

Como seguimiento a la respuesta de @StuartMarks . Mi biblioteca StreamEx tiene la takeWhileoperación que es compatible con la implementación actual de JDK-9. Cuando se ejecuta bajo JDK-9, simplemente delegará a la implementación de JDK (a través de la MethodHandle.invokeExactcual es realmente rápido). Cuando se ejecuta bajo JDK-8, se utilizará la implementación "polyfill". Entonces, usando mi biblioteca, el problema puede resolverse así:

IntStreamEx.iterate(1, n -> n + 1)
           .takeWhile(n -> n < 10)
           .forEach(System.out::println);
Tagir Valeev
fuente
¿Por qué no lo ha implementado para la clase StreamEx?
Someguy
@Someguy lo implementé.
Tagir Valeev
14

takeWhilees una de las funciones proporcionadas por la biblioteca 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
fuente
11

Actualización: Java 9 Streamahora viene con un método takeWhile .

No hay necesidad de hacks u otras soluciones. ¡Solo usa eso!


Estoy seguro de que esto se puede mejorar en gran medida: (alguien podría hacerlo seguro para subprocesos tal vez)

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 truco seguro ... No es elegante, pero funciona ~: 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);
    }

}
El coordinador
fuente
8

Puede usar 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));
frhack
fuente
6

En realidad, hay 2 formas de hacerlo en Java 8 sin bibliotecas adicionales o utilizando Java 9.

Si desea imprimir números del 2 al 20 en la consola, puede hacer esto:

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

o

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

La salida es en ambos casos:

2
4
6
8
10
12
14
16
18
20

Nadie mencionó anyMatch todavía. Esta es la razón de esta publicación.

gil.fernandes
fuente
5

Esta es la fuente copiada de JDK 9 java.util.stream.Stream.takeWhile (Predicate). Una pequeña diferencia para trabajar con 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);
}
marciano
fuente
4

Aquí hay una versión realizada en ints, como se hizo en la pregunta.

Uso:

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

Aquí hay código para 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
fuente
2

Ir a la biblioteca AbacusUtil . Proporciona la API exacta que desea y más:

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

Declaración: Soy el desarrollador de AbacusUtil.

user_3380739
fuente
0

No puede abortar una transmisión excepto por una operación de terminal de cortocircuito, lo que dejaría algunos valores de transmisión sin procesar independientemente de su valor. Pero si solo desea evitar operaciones en una secuencia, puede agregar una transformación y filtro a la secuencia:

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

Eso transforma el flujo de cosas en nulos cuando las cosas cumplen alguna condición, luego filtra los nulos. Si está dispuesto a disfrutar de los efectos secundarios, puede establecer el valor de la condición en verdadero una vez que se encuentre algo, para que todas las cosas posteriores se filtren independientemente de su valor. Pero incluso si no, puede guardar una gran cantidad de procesamiento (si no todo) al filtrar los valores fuera de la secuencia que no desea procesar.

Mateo
fuente
Es lamentable que algún evaluador anónimo rebajara mi respuesta sin decir por qué. Entonces, ni yo ni ningún otro lector sabemos qué hay de malo en mi respuesta. En ausencia de su justificación, consideraré que sus críticas no son válidas, y mi respuesta tal como se publicó es correcta.
Mateo
Su respuesta no resuelve el problema de los OP, que se trata de flujos infinitos. También parece complicar innecesariamente las cosas, ya que puede escribir la condición en la llamada filter (), sin necesidad de map (). La pregunta ya tiene un código de ejemplo, solo intente aplicar su respuesta a ese código y verá que el programa se repetirá para siempre.
SenoCtar
0

Incluso tenía un requisito similar: invocar el servicio web, si falla, vuelva a intentarlo 3 veces. Si falla incluso después de estas muchas pruebas, envíe una notificación por correo electrónico. Después de googlear mucho, anyMatch()vino como salvador. Mi código de muestra de la siguiente manera. En el siguiente ejemplo, si el método webServiceCall devuelve verdadero en la primera iteración en sí, stream no itera más como lo hemos llamado anyMatch(). Creo que esto es lo que estás buscando.

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
fuente
0

Si conoce la cantidad exacta de repeticiones que se realizarán, puede hacer

IntStream
          .iterate(1, n -> n + 1)
          .limit(10)
          .forEach(System.out::println);
Dilip Tharoor
fuente
1
Si bien esto podría responder a la pregunta de los autores, carece de algunas palabras explicativas y enlaces a la documentación. Los fragmentos de código sin formato no son muy útiles sin algunas frases a su alrededor. También puede encontrar cómo escribir una buena respuesta muy útil. Por favor edite su respuesta.
hola
0
    IntStream.iterate(1, n -> n + 1)
    .peek(System.out::println) //it will be executed 9 times
    .filter(n->n>=9)
    .findAny();

en lugar del pico, puede usar mapToObj para devolver el objeto o mensaje 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 Potomkin
fuente
-2

Si tiene un problema diferente, puede ser necesaria una solución diferente, pero para su problema actual, simplemente iría con:

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

Podría estar un poco fuera de tema, pero esto es para lo que tenemos List<T>más que para Stream<T>.

Primero necesitas tener un takemétodo util. Este método toma los primeros nelementos:

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

simplemente funciona como 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));

ahora será bastante simple escribir un takeWhilemétodo basado entake

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
}

funciona así:

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

esta implementación itera la lista parcialmente por algunas veces pero no agregará O(n^2)operaciones de agregar . Espero que sea aceptable.

Max
fuente
-3

Tengo otra solución rápida al implementar esto (que de hecho es muy inmundo, pero ya entiendes la idea):

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
fuente
2
¡Estás evaluando toda la transmisión por adelantado! Y si currentnunca .equals(e), obtendrás un bucle sin fin. Ambos, incluso si posteriormente solicita, por ejemplo .limit(1). Eso es mucho peor que 'inmundo' .
charlie
-3

Aquí está mi intento de usar solo la biblioteca Java Stream.

        IntStream.iterate(0, i -> i + 1)
        .filter(n -> {
                if (n < 10) {
                    System.out.println(n);
                    return false;
                } else {
                    return true;
                }
            })
        .findAny();
alpinismo
fuente
Se filtersupone que el predicado no tiene estado. System.out.printlnEs un efecto secundario.
Radiodef