Какие знаешь горячие потоки?

Горячие потоки в реактивном программировании

В контексте реактивного программирования (особенно с RxJava, RxKotlin, Flow API из Coroutines) горячие потоки (hot streams/observables) — это источники данных, которые начинают испускать элементы независимо от наличия подписчиков. Они активны сразу после создания, и данные могут быть потеряны, если подписка произошла позже начала эмиссии. Это фундаментальное отличие от холодных потоков, которые начинают работу только при подписке и для каждого подписчика создают независимый поток данных.

Ключевые характеристики горячих потоков

• Независимость от подписчиков: Эмиссия данных начинается сразу, даже если подписчиков нет.
• Общий источник данных: Все подписчики получают одни и те же данные, начиная с момента подписки (или все данные, если используется кэширование).
• Потенциальная потеря данных: Если подписчик подключился после начала эмиссии, он может пропустить часть данных.
• Часто требуют управления ресурсами: Так как они активны независимо от подписок, нужно явно управлять их жизненным циклом (запускать/останавливать).

Примеры горячих потоков в Android-разработке

1. Subjects в RxJava/RxKotlin

Subject действует одновременно как Observable и Observer, что делает его горячим по умолчанию:

val subject = PublishSubject.create<String>()

// Эмиссия начинается ДО подписки
subject.onNext("Event 1")
subject.onNext("Event 2")

// Подписчик получит только события, отправленные ПОСЛЕ подписки
subject.subscribe { event ->
    println("Received: $event") // Выведет только "Event 3"
}

subject.onNext("Event 3")

Основные типы Subjects:

• PublishSubject: Отправляет только новые события подписчикам.
• BehaviorSubject: Кэширует последнее значение и отправляет его новым подписчикам.
• ReplaySubject: Кэширует указанное количество событий и отправляет их новым подписчикам.
• AsyncSubject: Отправляет только последнее значение при завершении.

2. StateFlow и SharedFlow в Coroutines

В Kotlin Coroutines эти реализации специально созданы для горячих потоков:

// StateFlow - всегда имеет начальное значение и кэширует последнее
private val _uiState = MutableStateFlow<UiState>(UiState.Loading)
val uiState: StateFlow<UiState> = _uiState.asStateFlow()

// SharedFlow - более гибкий, можно настроить replay кэш
private val _events = MutableSharedFlow<String>(
    replay = 2, // Кэширует 2 последних события для новых подписчиков
    extraBufferCapacity = 10
)
val events: SharedFlow<String> = _events.asSharedFlow()

3. Broadcast Channel (устаревший, но исторически важный)

val channel = BroadcastChannel<String>(Channel.CONFLATED)

// Данные можно отправлять без подписчиков
channel.send("Message 1")

// Подписчик открывает свой receive-канал
val subscription = channel.openSubscription()

4. LiveData от Android Architecture Components

Хотя LiveData не является чисто реактивным в смысле Rx, он демонстрирует характеристики горячего потока:

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

Практическое применение в Android

1. Кеширование состояния приложения:

   // Глобальный StateFlow для авторизации
   object AuthManager {
       private val _authState = MutableStateFlow<AuthState>(AuthState.Unauthorized)
       val authState: StateFlow<AuthState> = _authState.asStateFlow()
       
       fun login(user: User) {
           _authState.value = AuthState.Authorized(user)
       }
   }
   
   // В разных компонентах приложения
   class ProfileFragment : Fragment() {
       override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
           lifecycleScope.launch {
               AuthManager.authState.collect { state ->
                   updateUI(state) // Всегда получим текущее состояние
               }
           }
       }
   }
   

2. Шина событий:

   object EventBus {
       private val _events = MutableSharedFlow<AppEvent>()
       val events = _events.asSharedFlow()
       
       suspend fun sendEvent(event: AppEvent) {
           _events.emit(event)
       }
   }
   
   // Отправка события из любого места
   lifecycleScope.launch {
       EventBus.sendEvent(NavigationEvent.OpenProfile)
   }
   

3. Дата-слои и репозитории:

   class UserRepository {
       private val _users = MutableStateFlow<List<User>>(emptyList())
       val users: StateFlow<List<User>> = _users.asStateFlow()
       
       init {
           viewModelScope.launch {
               loadUsersPeriodically() // Начинаем загрузку независимо от подписок
           }
       }
   }
   

Особенности управления жизненным циклом

Горячие потоки требуют внимательного подхода к управлению ресурсами:

• StateFlow автоматически поддерживает активную подписку, пока есть коллекторы
• В RxJava нужно явно управлять Disposable
• Важно отменять подписки в onDestroy или с помощью LifecycleScope

Сравнение с холодными потоками

Горячие потоки особенно полезны в Android для:

• Хранения и распространения состояния UI
• Реализации шин событий
• Кэширования данных
• Сценариев, где источник данных существует независимо от потребителей

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