«1. Обзор

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

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

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

2. Stream.takeWhile() из Java 9

Предположим, у нас есть поток элементов 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, используя цикл for и оператор break, чтобы помочь нам увидеть, как это работает: 

List<String> 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.spliterator. Мы можем заставить этот Spliterator выполнять перерыв для нас.

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

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

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

public class CustomSpliterator<T> extends Spliterators.AbstractSpliterator<T> {

    private Spliterator<T> splitr;
    private Predicate<T> predicate;
    private boolean isMatched = true;

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

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

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

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

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

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

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

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

4. Пользовательский forEach

Хотя предоставление Stream со встроенным механизмом прерывания может быть полезным, может быть проще сосредоточиться только на операции 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 <T> void forEach(Stream<T> stream, BiConsumer<T, Breaker> consumer) {
        Spliterator<T> spliterator = stream.spliterator();
        boolean hadNext = true;
        Breaker breaker = new Breaker();

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

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

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

@Test
public void whenCustomForEachIsCalled_ThenCorrectItemsAreReturned() {
    Stream<String> initialStream = Stream.of("cat", "dog", "elephant", "fox", "rabbit", "duck");
    List<String> 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 для потока. Мы увидели, как takeWhile в Java 9 решает большую часть проблемы за нас, и как предоставить версию этого для Java 8. поток.

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

«