Как вырваться из потока Java для каждого

1. Обзор

Как разработчики Java, мы часто пишем код, который повторяет набор элементов и выполняет операцию над каждым из них. Библиотека потоков Java 8 и ее для каждого метода позволяют нам писать этот код чистым, декларативным способом.

Хотя это похоже на циклы, нам не хватает эквивалента оператора break для прерывания итерации . Поток может быть очень длинным или потенциально бесконечным , и если у нас нет причин продолжать его обработку , мы хотели бы прервать его, а не ждать его последнего элемента.

В этом уроке мы рассмотрим некоторые механизмы, которые позволяют нам имитировать прерывание оператора в потоке.для каждой операции.

2. Поток Java 9.takeWhile()

Предположим, у нас есть поток String элементов, и мы хотим обрабатывать его элементы до тех пор, пока их длина нечетна.

Давайте попробуем Java 9 Stream.takeWhile метод:

Stream.of("cat", "dog", "elephant", "fox", "rabbit", "duck")
  .takeWhile(n -> n.length() % 2 != 0)
  .forEach(System.out::println);

Если мы запустим это, мы получим результат:

cat
dog

Давайте сравним это с эквивалентным кодом на обычной Java, использующим для цикла и прерывания оператора, чтобы помочь нам понять, как это работает:

List list = asList("cat", "dog", "elephant", "fox", "rabbit", "duck");
for (int i = 0; i < list.size(); i++) {
    String item = list.get(i);
    if (item.length() % 2 == 0) {
        break;
    }
    System.out.println(item);
}

Как мы видим, метод takeWhile позволяет нам достичь именно того, что нам нужно.

Но что, если мы еще не приняли Java 9? Как мы можем добиться аналогичного с помощью Java 8?

3. Пользовательский Разделитель

Давайте создадим пользовательский разделитель , который будет работать как декоратор для Stream.разделитель . Мы можем заставить этот Разделитель выполнить разрыв для нас.

Сначала мы получим Разделитель из нашего потока, затем украсим его нашим Пользовательским разделителем и предоставим Предикат для управления операцией break . Наконец, мы создадим новый поток из Пользовательского разделителя:

public static  Stream takeWhile(Stream stream, Predicate predicate) {
    CustomSpliterator customSpliterator = new CustomSpliterator<>(stream.spliterator(), predicate);
    return StreamSupport.stream(customSpliterator, false);
}

Давайте рассмотрим, как создать Пользовательский разделитель :

public class CustomSpliterator extends Spliterators.AbstractSpliterator {

    private Spliterator splitr;
    private Predicate predicate;
    private boolean isMatched = true;

    public CustomSpliterator(Spliterator splitr, Predicate predicate) {
        super(splitr.estimateSize(), 0);
        this.splitr = splitr;
        this.predicate = predicate;
    }

    @Override
    public synchronized boolean tryAdvance(Consumer consumer) {
        boolean hadNext = splitr.tryAdvance(elem -> {
            if (predicate.test(elem) && isMatched) {
                consumer.accept(elem);
            } else {
                isMatched = false;
            }
        });
        return hadNext && isMatched;
    }
}

Итак, давайте взглянем на метод tryAdvance . Здесь мы видим, что пользовательский Разделитель обрабатывает элементы оформленного Разделителя . Обработка выполняется до тех пор, пока наш предикат сопоставлен и в исходном потоке все еще есть элементы. Когда любое из условий становится ложным , наш Разделитель | “прерывается” и потоковая операция заканчивается.

Давайте протестируем наш новый вспомогательный метод:

@Test
public void whenCustomTakeWhileIsCalled_ThenCorrectItemsAreReturned() {
    Stream initialStream = 
      Stream.of("cat", "dog", "elephant", "fox", "rabbit", "duck");

    List result = 
      CustomTakeWhile.takeWhile(initialStream, x -> x.length() % 2 != 0)
        .collect(Collectors.toList());

    assertEquals(asList("cat", "dog"), result);
}

Как мы видим, поток прекратился после выполнения условия. В целях тестирования мы собрали результаты в список, но мы также могли бы использовать вызов forEach или любую другую функцию Stream .

4. Пользовательский интерфейс

Хотя предоставление потока со встроенным механизмом break может быть полезным, может быть проще сосредоточиться только на операции forEach .

Давайте использовать Stream.spliterator напрямую без декоратора:

public class CustomForEach {

    public static class Breaker {
        private boolean shouldBreak = false;

        public void stop() {
            shouldBreak = true;
        }

        boolean get() {
            return shouldBreak;
        }
    }

    public static  void forEach(Stream stream, BiConsumer consumer) {
        Spliterator spliterator = stream.spliterator();
        boolean hadNext = true;
        Breaker breaker = new Breaker();

        while (hadNext && !breaker.get()) {
            hadNext = spliterator.tryAdvance(elem -> {
                consumer.accept(elem, breaker);
            });
        }
    }
}

Как мы видим, новый пользовательский метод forEach вызывает Двоичный пользователь , предоставляя нашему коду как следующий элемент, так и объект прерывателя, который он может использовать для остановки потока.

Давайте попробуем это в модульном тесте:

@Test
public void whenCustomForEachIsCalled_ThenCorrectItemsAreReturned() {
    Stream initialStream = Stream.of("cat", "dog", "elephant", "fox", "rabbit", "duck");
    List result = new ArrayList<>();

    CustomForEach.forEach(initialStream, (elem, breaker) -> {
        if (elem.length() % 2 == 0) {
            breaker.stop();
        } else {
            result.add(elem);
        }
    });

    assertEquals(asList("cat", "dog"), result);
}

5. Заключение

В этой статье мы рассмотрели способы обеспечения эквивалента вызова break в потоке. Мы видели, как Java 9 takeWhile решает большую часть проблемы для нас и как предоставить версию этого для Java 8.

Наконец, мы рассмотрели служебный метод, который может предоставить нам эквивалент операции break при повторении потока .

Как всегда, пример кода можно найти на GitHub .