В корутинах поток – это тип, который может последовательно выдавать несколько значений, в отличие от функций приостановки, которые возвращают только одно значение. Например, с помощью потока можно получать обновления из базы данных в реальном времени.
Потоки создаются на основе сопрограмм и могут предоставлять несколько значений.
Поток – это поток данных, который можно обрабатывать асинхронно. Все значения должны быть одного типа. Например, Flow<Int> – это поток, который выдает целочисленные значения.
Поток очень похож на Iterator, который создает последовательность значений, но для создания и использования значений асинхронно он использует функции приостановки. Это означает, что, например, поток может безопасно выполнить сетевой запрос, чтобы получить следующее значение, не блокируя основной поток.
В потоках данных участвуют три объекта:
- Производитель создает данные, которые добавляются в поток. Благодаря сопрограммам потоки также могут создавать данные асинхронно.
- (Необязательно.) Посредники могут изменять каждое значение, передаваемое в поток, или сам поток.
- Потребитель использует значения из потока.
В Android репозиторий обычно передает данные интерфейса из локальных баз данных (Room) или облачных серверных частей (например, Firebase), а интерфейс выступает в качестве потребителя данных. И наоборот, интерфейс может создавать события ввода пользователя, которые будут использоваться другими уровнями. Промежуточные слои (например, экземпляры ViewModel) преобразуют поток в соответствии с требованиями каждого слоя.
Как создать процесс
Чтобы создавать процессы, используйте API конструктора процессов. Функция flow создает новый поток, в который можно вручную добавлять новые значения с помощью функции emit.
В следующем примере источник данных автоматически получает последние новости через заданные интервалы. Поскольку функция приостановки не может возвращать несколько последовательных значений, источник данных создает и возвращает поток, чтобы выполнить это требование. В этом случае источник данных выступает в роли производителя.
class NewsRemoteDataSource( private val newsApi: NewsApi, private val refreshIntervalMs: Long = 5000 ) { val latestNews: Flow<List<ArticleHeadline>> = flow { while (true) { val latestNews = newsApi.fetchLatestNews() emit(latestNews) // Emits the result of the request to the flow delay(refreshIntervalMs) // Suspends the coroutine for some time } } } // Interface that provides a way to make network requests with suspend functions interface NewsApi { suspend fun fetchLatestNews(): List<ArticleHeadline> }
Конструктор flow выполняется в рамках сопрограммы. Таким образом, он использует те же асинхронные API, но с некоторыми ограничениями:
- Процессы последовательны. Поскольку производитель находится в сопрограмме, при вызове функции приостановки он приостанавливается до тех пор, пока функция приостановки не вернет значение. В примере продюсер приостанавливает работу до тех пор, пока не будет выполнен сетевой запрос
fetchLatestNews. и только после этого результат передается в поток. - С помощью конструктора
flowпроизводитель не можетemitзначения из другогоCoroutineContext. Поэтому не вызывайтеemitв другомCoroutineContext, создавая новые сопрограммы или используя блоки кодаwithContext. В таких случаях можно использовать другие конструкторы, напримерcallbackFlow.
Как изменить трансляцию
Промежуточные операторы позволяют посредникам изменять поток данных, не используя значения. Эти операторы – функции, которые при применении к потоку данных создают цепочку операций, выполняемых только после того, как значения будут использованы. Подробнее о промежуточных операторах…
В примере ниже уровень хранилища использует промежуточный оператор map для преобразования данных, которые будут показаны на View:
class NewsRepository( private val newsRemoteDataSource: NewsRemoteDataSource, private val userData: UserData ) { /** * Returns the favorite latest news applying transformations on the flow. * These operations are lazy and don't trigger the flow. They just transform * the current value emitted by the flow at that point in time. */ val favoriteLatestNews: Flow<List<ArticleHeadline>> = newsRemoteDataSource.latestNews // Intermediate operation to filter the list of favorite topics .map { news -> news.filter { userData.isFavoriteTopic(it) } } // Intermediate operation to save the latest news in the cache .onEach { news -> saveInCache(news) } }
Промежуточные операторы можно применять один за другим, формируя цепочку операций, которые выполняются отложенно, когда элемент передается в поток. Обратите внимание, что простое применение промежуточного оператора к потоку не запускает сбор данных о потоке.
Как собирать данные из потока
Используйте терминальный оператор, чтобы запустить рабочий процесс и начать отслеживать значения. Чтобы получить все значения в потоке по мере их появления, используйте collect.
Подробнее о терминальных операторах можно узнать из официальной документации по потокам.
Поскольку collect является функцией приостановки, ее необходимо выполнять в рамках сопрограммы. В качестве параметра принимает лямбда-функцию, которая вызывается для каждого нового значения. Поскольку это функция приостановки, сопрограмма, вызывающая collect, может приостанавливаться до закрытия потока.
Ниже приведен простой пример реализации ViewModel, использующего данные из уровня хранилища, на основе предыдущего примера:
class LatestNewsViewModel( private val newsRepository: NewsRepository ) : ViewModel() { init { viewModelScope.launch { // Trigger the flow and consume its elements using collect newsRepository.favoriteLatestNews.collect { favoriteNews -> // Update UI with the latest favorite news } } } }
Сбор данных запускает продюсер, который обновляет последние новости и отправляет результат сетевого запроса через фиксированные интервалы. Поскольку производитель всегда активен в цикле while(true), поток данных будет закрыт, когда ViewModel будет очищена, а viewModelScope отменен.
Сбор данных о потоках может быть остановлен по следующим причинам:
- Сопрограмма, которая собирает данные, отменяется, как показано в предыдущем примере. Это также останавливает базовый источник.
- Продюсер завершает трансляцию элементов. В этом случае поток данных закрывается, а сопрограмма, вызвавшая
collect, возобновляет выполнение.
Потоки являются холодными и ленивыми, если не указаны другие промежуточные операторы. Это означает, что код производителя выполняется каждый раз, когда в потоке вызывается оператор терминала. В предыдущем примере наличие нескольких коллекторов потоков приводит к тому, что источник данных получает последние новости несколько раз с разными фиксированными интервалами. Чтобы оптимизировать и передать поток, когда несколько потребителей собирают данные одновременно, используйте оператор shareIn.
Как перехватывать неожиданные исключения
Реализация производителя может быть взята из сторонней библиотеки.
Это означает, что он может вызывать неожиданные исключения. Чтобы обрабатывать такие исключения, используйте промежуточный оператор catch.
class LatestNewsViewModel( private val newsRepository: NewsRepository ) : ViewModel() { init { viewModelScope.launch { newsRepository.favoriteLatestNews // Intermediate catch operator. If an exception is thrown, // catch and update the UI .catch { exception -> notifyError(exception) } .collect { favoriteNews -> // Update UI with the latest favorite news } } } }
В предыдущем примере при возникновении исключения лямбда-функция collect не вызывается, поскольку новый элемент не получен.
catch также может emit элементы в поток. В примере ниже уровень репозитория может emit кешированные значения:
class NewsRepository( // ... ) { val favoriteLatestNews: Flow<List<ArticleHeadline>> = newsRemoteDataSource.latestNews .map { news -> news.filter { userData.isFavoriteTopic(it) } } .onEach { news -> saveInCache(news) } // If an error happens, emit the last cached values .catch { exception -> emit(lastCachedNews()) } }
В этом примере при возникновении исключения вызывается лямбда-функция collect, поскольку из-за исключения в поток был добавлен новый элемент.
Выполнение в другом CoroutineContext
По умолчанию производитель сборщика flow выполняется в CoroutineContext сопрограммы, которая собирает данные из него, и, как уже упоминалось, он не может emit значения из другого CoroutineContext. В некоторых случаях такое поведение может быть нежелательным.
Например, в примерах, приведенных в этой статье, уровень репозитория не должен выполнять операции с Dispatchers.Main, используемым viewModelScope.
Чтобы изменить CoroutineContext потока, используйте промежуточный оператор flowOn.
flowOn изменяет CoroutineContext вышестоящего потока, то есть производителя и любые промежуточные операторы, примененные до (или выше)
flowOn. Последующий поток (промежуточные операторы после flowOn
вместе с потребителем) не затрагивается и выполняется на CoroutineContext, используемом для collect из потока. Если операторов flowOn несколько, каждый из них меняет поток данных, поступающий из источника, расположенного выше по течению.
class NewsRepository( private val newsRemoteDataSource: NewsRemoteDataSource, private val userData: UserData, private val defaultDispatcher: CoroutineDispatcher ) { val favoriteLatestNews: Flow<List<ArticleHeadline>> = newsRemoteDataSource.latestNews .map { news -> // Executes on the default dispatcher news.filter { userData.isFavoriteTopic(it) } } .onEach { news -> // Executes on the default dispatcher saveInCache(news) } // flowOn affects the upstream flow ↑ .flowOn(defaultDispatcher) // the downstream flow ↓ is not affected .catch { exception -> // Executes in the consumer's context emit(lastCachedNews()) } }
В этом коде операторы onEach и map используют defaultDispatcher, а оператор catch и потребитель выполняются на Dispatchers.Main, используемом viewModelScope.
Поскольку уровень источника данных выполняет операции ввода-вывода, вам следует использовать диспетчер, оптимизированный для таких операций:
class NewsRemoteDataSource( // ... private val ioDispatcher: CoroutineDispatcher ) { val latestNews: Flow<List<ArticleHeadline>> = flow { // Executes on the IO dispatcher // ... } .flowOn(ioDispatcher) }
Потоки в библиотеках Jetpack и источниках данных
Flow интегрирован во многие библиотеки Jetpack и популярен в сторонних библиотеках Android и облачных SDK. Flow отлично подходит для обновления данных в реальном времени и непрерывной передачи данных.
Вы можете использовать Flow with Room, чтобы получать уведомления об изменениях в базе данных. При использовании объектов доступа к данным (DAO) возвращайте тип Flow, чтобы получать обновления в реальном времени.
@Dao abstract class ExampleDao { @Query("SELECT * FROM Example") abstract fun getExamples(): Flow<List<Example>> }
Каждый раз, когда в таблице Example происходят изменения, создается новый список с новыми элементами в базе данных.
Аналогичным образом источники облачных данных в реальном времени, такие как Cloud Firestore, предоставляют
Flow расширения, которые передают обновления данных в реальном времени непосредственно в слой
репозитория:
class RemoteMessagesDataSource(
private val firestore: FirebaseFirestore
) {
// Returns a Flow emitting updated message lists when collection changes
val messages: Flow<List<Message>> =
firestore.collection("messages").dataObjects()
}
Это хорошо сочетается с локальным хранилищем: ваш репозиторий может объединять потоки из Room (локальный кеш) и Firebase (удаленные облачные данные), чтобы обеспечить работу в режиме офлайн и в реальном времени.
Как преобразовать API на основе обратного вызова в визуальные сценарии
callbackFlow – это конструктор потоков, который позволяет преобразовывать API на основе обратного вызова в потоки. Это особенно полезно при работе с несколькими прослушивателями событий в SDK, например Cloud Firestore или Firebase Remote Config.
Чтобы преобразовать API на основе обратного вызова в поток (например, для отслеживания обновлений базы данных Firestore), используйте следующий код:
class FirestoreUserEventsDataSource(
private val firestore: FirebaseFirestore
) {
// Method to get user events from the Firestore database
fun getUserEvents(): Flow<UserEvents> = callbackFlow {
// Reference to use in Firestore
var eventsCollection: CollectionReference? = null
try {
eventsCollection = FirebaseFirestore.getInstance()
.collection("collection")
.document("app")
} catch (e: Throwable) {
// If Firebase cannot be initialized, close the stream of data
// flow consumers will stop collecting and the coroutine will resume
close(e)
}
// Registers callback to firestore, which will be called on new events
val subscription = eventsCollection?.addSnapshotListener { snapshot, _ ->
if (snapshot == null) { return@addSnapshotListener }
// Sends events to the flow! Consumers will get the new events
try {
trySend(snapshot.getEvents())
} catch (e: Throwable) {
// Event couldn't be sent to the flow
}
}
// The callback inside awaitClose will be executed when the flow is
// either closed or cancelled.
// In this case, remove the callback from Firestore
awaitClose { subscription?.remove() }
}
}
В отличие от конструктора flow, функция callbackFlow позволяет получать значения из другой функции CoroutineContext с помощью функции send или вне сопрограммы с помощью функции trySend.
Внутренне callbackFlow использует канал, который концептуально очень похож на блокирующую очередь.
Канал настраивается с помощью емкости – максимального количества элементов, которые могут быть буферизованы. Канал, созданный в callbackFlow, по умолчанию может содержать до 64 элементов. Если вы попытаетесь добавить новый элемент на полный канал, send приостановит работу производителя, пока не освободится место для нового элемента, а trySend не добавит элемент на канал и сразу вернет false.
trySend немедленно добавляет указанный элемент в канал, только если это не нарушает ограничений на емкость, а затем возвращает успешный результат.