0Pricing
Kotlin Academy · Урок

Преобразование холодного Flow в горячий с shareIn и stateIn

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

«Преобразование холодного Flow в горячий с shareIn и stateIn» — бесплатный урок Kotlin Academy на CoddyKit. Это урок 3 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения Kotlin Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс Kotlin Academy содержит 4 уроков всего.

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

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

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun expensiveFlow() = flow {
    println("Starting upstream work")  // runs once if shared
    repeat(3) { delay(100); emit(it) }
}
fun main() = runBlocking {
    val cold = expensiveFlow()
    // Two collectors = two executions:
    launch { cold.collect { } }
    launch { cold.collect { } }
    delay(500)
    coroutineContext.cancelChildren()
}

Основы shareIn

flow.shareIn(scope, started, replay) преобразует холодный Flow в SharedFlow, используя одну общую подписку на исходный поток.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val cold = flow {
        println("Upstream started once")
        repeat(5) { delay(100); emit(it) }
    }
    val hot = cold.shareIn(this, SharingStarted.Eagerly, replay = 0)
    launch { hot.collect { println("A: $it") } }
    launch { hot.collect { println("B: $it") } }
    delay(600)
    coroutineContext.cancelChildren()
}

Основы stateIn

flow.stateIn(scope, started, initialValue) преобразует поток в StateFlow — он всегда имеет значение и повторно выдаёт одно последнее значение.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val cold = flow { delay(100); emit(42) }
    val state: StateFlow<Int> = cold.stateIn(
        scope = this,
        started = SharingStarted.Eagerly,
        initialValue = 0
    )
    println(state.value)  // 0 immediately
    delay(200)
    println(state.value)  // 42 after upstream emits
    coroutineContext.cancelChildren()
}

Стратегия SharingStarted с немедленным запуском

С немедленным запуском: исходный поток запускается сразу при вызове shareIn/stateIn, независимо от наличия подписчиков.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val flow = flow {
        println("Eagerly started")
        emit(1)
    }.shareIn(this, SharingStarted.Eagerly)
    // Upstream already running even before any collect
    delay(50)
    launch { flow.collect { println(it) } }
    delay(100)
    coroutineContext.cancelChildren()
}

Отложенная стратегия SharingStarted

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

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val flow = flow {
        println("Lazily started on first subscriber")
        repeat(3) { delay(100); emit(it) }
    }.shareIn(this, SharingStarted.Lazily, replay = 1)
    delay(50)  // no subscriber yet — not started
    launch { flow.collect { println(it) } }  // triggers start
    delay(400)
    coroutineContext.cancelChildren()
}

SharingStarted.WhileSubscribed

WhileSubscribed(stopTimeout): исходный поток запускается при подключении первого подписчика и останавливается через stopTimeout мс после ухода последнего подписчика. Идеальный вариант для ViewModels.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val flow = flow {
        println("Started")
        repeat(10) { delay(100); emit(it) }
    }.shareIn(this, SharingStarted.WhileSubscribed(500))
    val job = launch { flow.collect { print("$it ") } }
    delay(300)
    job.cancel()    // subscriber left
    delay(200)      // within 500ms stop timeout — still running
    launch { flow.collect { print("resume $it ") } }
    delay(500)
    coroutineContext.cancelChildren()
}

Типичный шаблон ViewModel

В Android ViewModels преобразуйте потоки репозитория с помощью stateIn(viewModelScope, WhileSubscribed(5000), initialValue), чтобы делиться ими и сохранять их при изменениях конфигурации.

import kotlinx.coroutines.flow.*
// In ViewModel:
// val uiState: StateFlow<UiState> = repository
//     .dataFlow()
//     .map { UiState.Success(it) }
//     .stateIn(
//         scope = viewModelScope,
//         started = SharingStarted.WhileSubscribed(5_000),
//         initialValue = UiState.Loading
//     )
fun main() { println("WhileSubscribed(5000) is the recommended ViewModel pattern") }

Сравнение shareIn и stateIn

shareIn → SharedFlow (без текущего значения, повторную выдачу можно настроить). stateIn → StateFlow (всегда имеет значение, повторная выдача одного значения). Выбирайте вариант в зависимости от того, нужно ли потребителям текущее состояние.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val cold = (1..3).asFlow().map { it * 10 }

    // SharedFlow — no initial value:
    val shared: SharedFlow<Int> = cold.shareIn(this, SharingStarted.Eagerly, replay = 1)

    // StateFlow — always has a value:
    val state: StateFlow<Int> = cold.stateIn(this, SharingStarted.Eagerly, 0)

    delay(100)
    println("shared cache: ${shared.replayCache}")
    println("state value: ${state.value}")
    coroutineContext.cancelChildren()
}

Компромиссы кэша повторной выдачи

Чем больше значение повторной выдачи, тем больше истории видят поздние подписчики, но тем больше памяти используется. Для состояния интерфейса достаточно одного значения повторной выдачи (stateIn). Для журналов событий может потребоваться большее значение.

import kotlinx.coroutines.flow.*
// replay=0: no history, only future events
// replay=1: last value (equivalent to stateIn)
// replay=N: last N events — use for message feeds, logs

// Trade-off: memory vs subscriber freshness
fun main() { println("Choose replay based on late-subscriber requirements") }

Риск отсутствия общей подписки на поток

Без общей подписки каждый сборщик Compose или наблюдатель ViewModel заново запускает исходный поток — это приводит к лишним сетевым вызовам и запросам к базе данных. Всегда используйте общую подписку для ресурсоёмких потоков.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun expensiveApi() = flow {
    println("API CALL")  // without sharing: once per collector
    emit("data")
}
fun main() = runBlocking {
    val shared = expensiveApi().shareIn(this, SharingStarted.Lazily, replay = 1)
    launch { shared.collect { } }  // one API call, shared
    launch { shared.collect { } }  // same emission
    delay(200)
    coroutineContext.cancelChildren()
}

Освобождение ресурсов при общей подписке

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

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun resourceFlow() = flow {
    try { repeat(10) { delay(100); emit(it) } }
    finally { println("Upstream cleaned up") }
}
fun main() = runBlocking {
    val scope = CoroutineScope(SupervisorJob())
    val shared = resourceFlow().shareIn(scope, SharingStarted.Eagerly)
    launch { shared.collect { print("$it ") } }
    delay(250)
    scope.cancel()  // upstream cleanup runs
    delay(100)
}

Быстрая проверка

Какая стратегия SharingStarted рекомендуется для Android ViewModels?

Повторение

shareIn преобразует поток в SharedFlow, а stateIn — в StateFlow. Используйте WhileSubscribed в ViewModels для экономии ресурсов. Общая подписка предотвращает лишние запуски исходного потока при подписке нескольких сборщиков.

Часто задаваемые вопросы

Урок «Преобразование холодного Flow в горячий с shareIn и stateIn» бесплатный?

Да — полный текст урока «Преобразование холодного Flow в горячий с shareIn и stateIn» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс Kotlin Academy, подпишись на CoddyKit PRO. Курс Kotlin Academy содержит 4 уроков всего.

Чему я научусь в уроке «Преобразование холодного Flow в горячий с shareIn и stateIn»?

Преобразуйте холодные Flow в горячие потоки с помощью операторов совместного использования. Ты практикуешь Kotlin Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать Kotlin Academy?

Предыдущий опыт не требуется. Kotlin Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 3 из 4.

Сколько времени занимает урок «Преобразование холодного Flow в горячий с shareIn и stateIn»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке Kotlin Academy?

Да. Каждый урок Kotlin Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

  1. StateFlow: горячий контейнер состояния для интерфейса
  2. SharedFlow: шины событий и одноразовые события
  3. Преобразование холодного Flow в горячий с shareIn и stateIn
  4. Тестирование StateFlow и SharedFlow с Turbine
← Назад к Kotlin Academy