Преобразование потоков

Akka Streams предоставляет богатый набор операторов для Source и Flow. Они напоминают API коллекций Scala, но работают асинхронно и соблюдают обратное давление.

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

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

Оба оператора сохраняют порядок элементов и передают вниз по потоку сигналы завершения и ошибки.

Если один входной элемент должен порождать несколько выходных, используйте mapConcat. Он принимает функцию, возвращающую итерируемую последовательность, и разворачивает результаты в поток.

Возврат пустой коллекции фактически удаляет элемент.

grouped(n) объединяет идущие подряд элементы в Seq размером не более n, что удобно для массовой записи в базу данных. sliding(n) выдаёт перекрывающиеся окна.

Пакетная обработка уменьшает накладные расходы на отдельный элемент в конвейерах с интенсивным вводом-выводом.

scan выдаёт текущее накопленное значение после каждого элемента, формируя изменяющийся поток состояния. fold выдаёт только итоговое накопленное значение после завершения вышестоящего этапа.

Используйте scan для текущих счётчиков, а fold — для итоговой агрегации.

mapAsync(parallelism) вызывает функцию, возвращающую Future, и выдаёт результаты в исходном порядке, одновременно выполняя до parallelism Future.

Используйте его для асинхронных вызовов, например обращений к базе данных или HTTP-запросов, когда порядок имеет значение.

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

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

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

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

Потоки можно преобразовывать не только по содержимому, но и по времени. throttle ограничивает скорость выдачи, groupedWithin объединяет элементы по размеру или истечении времени, а takeWithin ограничивает продолжительность.

Эти операторы незаменимы для ограничения частоты обращений к внешним API.

Исключение, возникшее внутри оператора, по умолчанию приводит к сбою всего потока. Стратегия надзора может вместо этого выполнить resume (удалить ошибочный элемент) или restart этапа.

Подключите стратегию с помощью withAttributes у Flow.

Небольшие Flow объединяются в более крупные с помощью via, образуя единый повторно используемый Flow. Это позволяет сосредоточить каждый этап на одной задаче и независимо тестировать его.

Объединённый Flow имеет входной тип первого этапа и выходной тип последнего.

Подумайте об асинхронных преобразованиях и гарантиях сохранения порядка.

Вы изучили операторы преобразования: поэлементные map/filter, преобразование одного элемента во множество с помощью mapConcat, группировку с помощью grouped, накопление с помощью scan и fold, а также асинхронную обработку через mapAsync.

Вы также рассмотрели преобразования с состоянием, операторы времени, такие как throttle, стратегии надзора для ошибок и объединение Flow с помощью via. Далее: как обратное давление обеспечивает безопасность этих этапов.