Лето — лучшее время для сплава. Поэтому, если вы пока не в отпуске, давайте устроим короткий сплав по асинхронным потокам данных.

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

Термины

Kotlin Flow — это API для работы с асинхронными потоками данных, построенное поверх корутин. Kotlin Flow реализует парадигму реактивного программирования.

Реактивное программирование — это парадигма, в которой все данные и события рассматриваются не как единичные значения, а как асинхронные потоки, протекающие через приложение. Каждый компонент «подписан» на поток входных событий и автоматически обновляет своё состояние в ответ на любые изменения.

Реактивное программирование можно рассматривать как синтез трех подходов:

Как и в декларативной парадигме, в реактивном программировании описывается «что», а не «как». Вместо явных циклов вы просто собираете цепочку операторов (map, filter, flatMap и т. д.), задавая желаемую трансформацию потока.

Как и в программировании потоков данных, в реактивном программировании каждый поток и каждый оператор образуют «узлы» и «ребра» графа, по которому данные протекают от источников к потребителям.

Как и в событийно-ориентированной парадигме, в реактивном программировании потоки часто реализуют модель публикации-подписки.

Простейший пример реактивной парадигмы мы все видели в Excel, где при обновлении значения ячейки автоматически пересчитываются все остальные ячейки, которые ссылаются на нее через формулы.

Холодные и горячие потоки

Поток данных (stream) — это последовательность значений, которые поступают со временем, зачастую асинхронно.

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

Горячие потоки генерируют данные независимо от наличия подписчиков, а холодные потоки ленивы и начинают генерировать их только по запросу.

Слева — «горячий поток» в виде извергающегося вулкана, непрерывно выбрасывающего данные (желтые листки), а справа — «холодный поток» в виде ледяного крана, из которого данные капают только при повороте вентиля.

«Холодные потоки, горячие каналы»

Так называется статья Романа Елизарова, опубликованная в 2019 году. Дело в том, что в Kotlin Flow есть и горячие (SharedFlow и StateFlow), и холодные (Flow) потоки данных. Кроме того в Kotlin есть еще одна реализация горячих потоков — Channels.

Чтобы не запутаться, давайте посмотрим на схему:

Объект Flow в Kotlin — это реализация холодного потока данных, который производит значения асинхронно и лениво. Таким образом, он берет лучшее и от корутин, и от последовательностей: значения вычисляются «на лету» и при этом используются корутины для асинхронной обработки данных.

За информацией о каналах отсылаю вас к своему курсу по корутинам на Степике. В данной статье мы будем говорить только о Flow и его вариациях.

Создание Flow

Хрестоматийный пример для объяснения работы Flow — вычисление чисел Фибоначчи:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun fibonacciFlow(): Flow<Long> = flow {
    var a = 0L
    var b = 1L
    emit(a)
    delay(100)
    emit(b)
    while (true) {
        val next = a + b
        emit(next)
        a = b
        b = next
        delay(100)
    }
}

fun main() = runBlocking {
    fibonacciFlow().take(10).collect { println(it) }
}

Здесь мы создаем поток данных с помощью билдера flow, получающего на вход лямбду с приемником. Как известно, если последний параметр функции в Kotlin сам является функцией, то ее можно в виде лямбда-выражения вынести за круглые скобки (trailing lambda). Поэтому вместо

flow() { ... }

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

Функция emit() отправляет очередное значение в поток, и оно плывет к подписчику — тому, кто запросил это значение. Так как emit() — это suspend-функция, она приостанавливает выполнение корутины после передачи каждого значения до следующего запроса.

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

Структура Flow

Каждый Flow делится на три части:

• эмиттер — блок, в котором генерируются данные и, как правило, вызывается функция emit() для отправки значений;

• промежуточные операторы — операторы, которые обрабатывают данные, проходящие через поток;

• коллектор — терминальный оператор, который запускает сбор потока и получает итоговые значения.

Взаимоотношения между терминальным и промежуточными операторами выглядят примерно так:

Возвращаясь к нашему примеру с числами Фибоначчи, эмиттер — это блок внутри flow{}:

    var a = 0L
    var b = 1L
    emit(a)
    delay(100)
    emit(b)
    while (true) {
        val next = a + b
        emit(next)
        a = b
        b = next
        delay(100)
    }

Промежуточный оператор представлен вызовом функции take():

.take(10)

А о коллекторе вы уже догадались по названию.

Многие промежуточные операторы называются так же, как и методы коллекций: map, filter, take и т.д. Однако они адаптированы для работы с асинхронными потоками данных и используют возможности корутин.

Например, создадим поток углов в градусах, затем последовательно применим преобразования с помощью оператора map:

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.math.*

fun main() = runBlocking {
    flowOf(0, 30, 45, 60, 90)
        .map { it.toDouble() } // преобразуем целые числа в Double
        .map { Math.toRadians(it) } // переводим градусы в радианы
        .map { sin(it) } // вычисляем синус угла
        .collect { println(it) } // выводим результат на экран
}

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

Как в случае и с коллекциями, подобные цепочки реализуют декларативную парадигму программирования: указано, что сделать, а не как сделать.

Пример

Предположим, у нас есть Android-приложение, в котором нам нужно отображать на главном экране актуальное количество записей из какой-то таблицы в БД. Актуальное — то есть при изменении таблицы это значение должно оперативно пересчитываться.

@Dao
interface WordDao {
  @Query("SELECT COUNT(id) FROM words")
  fun count(): Flow<Int>
}

С помощью Flow мы следим за таблицей words и автоматически эмиттим обновленный count при каждом изменении данных.

@Singleton
class WordRepository @Inject constructor(
  private val wordDao: WordDao
) {
  fun countWordsFlow(): Flow<Int> =
    wordDao.count().
      .distinctUntilChanged()
      .flowOn(Dispatchers.IO)
}

Метод countWordsFlow()позволяет подписаться на обновления количества слов в базе. Как только данные изменятся, в соответствующий Flow поступит новое значение — но только если оно реально изменилось (за счет вызова distinctUntilChanged). Все это выполняется асинхронно и не блокирует основной поток приложения. Поэтому нам не приходится каждый раз обращаться к базе данных вручную.

Итог

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

Больше теории и примеров — в моем курсе по корутинам.