Java 8 и бесконечные потоки

1. Обзор

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

Возможность работы с бесконечной последовательностью элементов основана на том факте, что потоки созданы ленивыми.

Эта ленивость достигается разделением двух типов операций, которые могут выполняться в потоках: промежуточные и терминальные операции.

2. Промежуточные и конечные операции

Все потоковые операции делятся на промежуточные и терминальные операции и объединяются в потоковые конвейеры.

Поточный конвейер состоит из источника (например, Collection , массива, функции-генератора, канала ввода-вывода или генератора бесконечной последовательности); за которыми следует ноль или более промежуточных операций и терминальная операция.

2.1. Промежуточные операции

Промежуточные операции не выполняются до тех пор, пока не будет вызвана какая-либо терминальная операция.

Они составлены, образуя конвейер выполнения Stream . Промежуточную операцию можно добавить в конвейер Stream методами:

• фильтр()
• карта()
• плоская карта ()
• отчетливый()
• отсортировано()
• заглянуть()
• лимит()
• пропускать()

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

По сути, промежуточные операции возвращают новый поток. Выполнение промежуточной операции фактически не выполняет никакой операции, а вместо этого создает новый поток, который при обходе содержит элементы исходного потока, соответствующие заданному предикату.

Таким образом, обход потока не начинается до тех пор, пока не будет выполнена терминальная операция конвейера.

Это очень важное свойство, особенно важное для бесконечных потоков, потому что оно позволяет нам создавать потоки, которые будут фактически вызываться только при вызове операции терминала .

2.2. Терминальные операции

Терминальные операции могут пересекать поток для получения результата или побочного эффекта.

После выполнения терминальной операции потоковый конвейер считается израсходованным и больше не может использоваться. Почти во всех случаях терминальные операции активны, завершая обход источника данных и обработку конвейера перед возвратом.

Оперативность терминальной операции важна в отношении бесконечных потоков, потому что в момент обработки нам нужно тщательно подумать, правильно ли наш поток ограничен , например, преобразованием limit() . Терминальные операции:

• для каждого()
• forEachOrdered()
• массив()
• уменьшать()
• собирать()
• мин()
• Максимум()
• считать()
• любое совпадение()
• все совпадения()
• нет совпадений ()
• найтиПервый()
• найтилюбой()

Каждая из этих операций инициирует выполнение всех промежуточных операций.

3. Бесконечные потоки

Теперь, когда мы понимаем эти две концепции — промежуточные и терминальные операции — мы можем написать бесконечный поток, который использует ленивость потоков.

Допустим, мы хотим создать бесконечный поток элементов с нуля, который будет увеличиваться на два. Затем нам нужно ограничить эту последовательность перед вызовом операции терминала.

Крайне важно использовать метод limit() перед выполнением метода collect () , который является терминальной операцией, иначе наша программа будет работать бесконечно:

// given
Stream<Integer> infiniteStream = Stream.iterate(0, i -> i + 2);

// when
List<Integer> collect = infiniteStream
  .limit(10)
  .collect(Collectors.toList());

// then
assertEquals(collect, Arrays.asList(0, 2, 4, 6, 8, 10, 12, 14, 16, 18));

Мы создали бесконечный поток, используя метод iterate() . Затем мы вызвали преобразование limit() и терминальную операцию collect() . Тогда в нашем результирующем списке у нас будут первые 10 элементов бесконечной последовательности из-за лени Stream.

4. Бесконечный поток пользовательского типа элементов

Допустим, мы хотим создать бесконечный поток случайных UUID .

Первым шагом к достижению этого с помощью Stream API является создание поставщика этих случайных значений:

Supplier<UUID> randomUUIDSupplier = UUID::randomUUID;

Когда мы определяем поставщика, мы можем создать бесконечный поток, используя метод generate() :

Stream<UUID> infiniteStreamOfRandomUUID = Stream.generate(randomUUIDSupplier);

Тогда мы могли бы взять пару элементов из этого потока. Нам нужно помнить об использовании метода limit() , если мы хотим, чтобы наша программа завершилась за конечное время:

List<UUID> randomInts = infiniteStreamOfRandomUUID
  .skip(10)
  .limit(10)
  .collect(Collectors.toList());

Мы используем преобразование skip() , чтобы отбросить первые 10 результатов и взять следующие 10 элементов. Мы можем создать бесконечный поток любых элементов пользовательского типа, передав функцию интерфейса Supplier методу generate() в Stream .

6. Do-While – путь потока

Допустим, у нас есть простой цикл do..while в нашем коде:

int i = 0;
while (i < 10) {
    System.out.println(i);
    i++;
}

Мы печатаем i counter десять раз. Мы можем ожидать, что такую конструкцию можно легко написать с помощью Stream API, и в идеале у нас должен быть метод doWhile() для потока.

К сожалению, в потоке такого метода нет, и когда мы хотим добиться функциональности, аналогичной стандартному циклу do- while, нам нужно использовать метод limit() :

Stream<Integer> integers = Stream
  .iterate(0, i -> i + 1);
integers
  .limit(10)
  .forEach(System.out::println);

Мы достигли той же функциональности, что и императивный цикл while, но с меньшим количеством кода, но вызов функции limit() не такой описательный, как если бы у нас был метод doWhile() для объекта Stream .

5. Вывод

В этой статье объясняется, как мы можем использовать Stream API для создания бесконечных потоков. Они, при использовании вместе с преобразованиями, такими как limit() , могут значительно упростить понимание и реализацию некоторых сценариев.

Код, поддерживающий все эти примеры, можно найти в проекте GitHub — это проект Maven, поэтому его легко импортировать и запускать как есть.