Запуск конвейера

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

Ничего не происходит, пока Вы явно не запустите граф, благодаря чему Akka Streams можно составлять и повторно использовать.

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

Обычно один ActorSystem обслуживает всё приложение и множество параллельных потоков.

Самый прямой способ запустить Source — использовать runWith: он подключает Sink и выполняет материализацию одним шагом, возвращая материализуемое значение этого Sink.

В данном случае результатом будет Future[Int], завершающийся суммой после окончания потока.

Если Вы уже создали замкнутый RunnableGraph с помощью to или toMat, вызовите run(), чтобы материализовать его. Возвращаемым значением будет то материализуемое значение, которое сохранил граф.

Так построение конвейера чётко отделяется от его выполнения.

Source предоставляет сокращённые варианты: runForeach, runFold и runReduce подключают соответствующий Sink и сразу запускают его.

Они удобны для распространённых конечных операций над Source.

Конечные Sink возвращают Future, который завершается при окончании потока или возникновении ошибки. Зарегистрируйте обратные вызовы с помощью onComplete, чтобы реагировать на успех или ошибку.

Используйте диспетчер ActorSystem как неявный ExecutionContext для этих обратных вызовов.

Типичный конвейер обработки данных читает данные из источника, преобразует их с помощью потоков, выполняет асинхронный ввод-вывод с помощью mapAsync, объединяет данные в пакеты с помощью grouped и записывает их в приемник.

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

Для повышения отказоустойчивости оберните источник или поток в RestartSource.withBackoff, чтобы временные сбои (например, разрыв соединения) автоматически запускали повторно с экспоненциально увеличивающейся задержкой.

Это позволяет поддерживать длительно работающие конвейеры загрузки данных без ручного управления.

KillSwitch позволяет внешнему коду корректно остановить работающий поток. Вставьте KillSwitches.single с помощью viaMat и сохраните его материализованное значение, чтобы позднее вызвать shutdown().

Это особенно важно для потоков, работающих в течение длительного времени и требующих остановки при завершении приложения.

При завершении приложения вызовите terminate для ActorSystem, чтобы освободить его потоки. Добавьте вызов terminate после Future завершения потока, чтобы завершение прошло упорядоченно.

Утечка ActorSystem не дает JVM завершить работу и удерживает ресурсы открытыми.

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

Это позволяет безопасно определить конвейер один раз и запускать его по требованию для каждой поступающей задачи.

Запуск конвейера означает материализацию схемы с помощью ActorSystem через run, runWith или удобные операторы; каждый из них возвращает результат типа Future.

Вы рассмотрели реалистичные многоэтапные конвейеры, автоматические перезапуски с задержкой, корректное завершение с помощью KillSwitch, освобождение ресурсов через system.terminate() и безопасное повторное использование неизменяемой схемы в независимых запусках.