Begrenzen Sie einen Stream durch ein Prädikat

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?

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

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/

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

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

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

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

}

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

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.

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

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

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.

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.

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

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

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

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

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.

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

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