做代码的网站,微信公众号登录二维码,福建建设厅网站官网,WORDPRESS摘要无效文章目录 一 什么是数据流二 创建数据流三 修改数据流四 从数据流中进行收集五 数据流捕获异常六 在不同 CoroutineContext 中执行七 Jetpack 库中的数据流八 将基于回调的 API 转换为数据流 一 什么是数据流
数据流以协程为基础构建#xff0c;可提供多个值。从概念上来讲可提供多个值。从概念上来讲数据流是可通过异步方式进行计算处理的一组数据序列
数据流包含三个实体
提供方会生成添加到数据流中的数据。得益于协程数据流还可以异步生成数据。可选中介可以修改发送到数据流的值或修正数据流本身。使用方则使用数据流中的值。
二 创建数据流
如需创建数据流请使用数据流构建器 API。flow 构建器函数会创建一个新数据流可使用 emit 函数手动将新值发送到数据流中。
class NewsRemoteDataSource(private val newsApi: NewsApi,private val refreshIntervalMs: Long 5000
) {val latestNews: FlowListArticleHeadline flow {while(true) {val latestNews newsApi.fetchLatestNews()emit(latestNews) // Emits the result of the request to the flowdelay(refreshIntervalMs) // Suspends the coroutine for some time}}
}// Interface that provides a way to make network requests with suspend functions
interface NewsApi {suspend fun fetchLatestNews(): ListArticleHeadline
}flow 构建器在协程内执行。因此它将受益于相同异步 API但也存在一些限制
数据流是有序的。当协程内的提供方调用挂起函数时提供方会挂起直到挂起函数返回。在此示例中提供方会挂起直到 fetchLatestNews 网络请求完成为止。只有这样请求结果才会发送到数据流中。使用 flow 构建器时提供方不能提供来自不同 CoroutineContext 的 emit 值。因此请勿通过创建新协程或使用 withContext 代码块在不同 CoroutineContext 中调用 emit。在这些情况下可使用其他数据流构建器例如 callbackFlow。
三 修改数据流
中介可以利用中间运算符如map在不使用值的情况下修改数据流。这些运算符都是函数可在应用于数据流时设置一系列暂不执行的链式运算留待将来使用值时执行。
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 dont trigger the flow. They just transform* the current value emitted by the flow at that point in time.*/val favoriteLatestNews: FlowListArticleHeadline 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
class LatestNewsViewModel(private val newsRepository: NewsRepository
) : ViewModel() {init {viewModelScope.launch {// Trigger the flow and consume its elements using collectnewsRepository.favoriteLatestNews.collect { favoriteNews -// Update View with the latest favorite news}}}
}数据流收集可能会由于以下原因而停止
如上例所示协程收集被取消。此操作也会让底层提供方停止活动。提供方完成发出数据项。在这种情况下数据流将关闭调用 collect 的协程则继续执行。
五 数据流捕获异常
使用 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 View with the latest favorite news}}}
}六 在不同 CoroutineContext 中执行
flow 构建器的提供方会通过从中收集的协程的 CoroutineContext 执行并且如前所述它无法从不同 CoroutineContext 对值执行 emit 操作。如需更改数据流的 CoroutineContext使用中间运算符 flowOn
class NewsRepository(private val newsRemoteDataSource: NewsRemoteDataSource,private val userData: UserData,private val defaultDispatcher: CoroutineDispatcher
) {val favoriteLatestNews: FlowListArticleHeadline newsRemoteDataSource.latestNews.map { news - // Executes on the default dispatchernews.filter { userData.isFavoriteTopic(it) }}.onEach { news - // Executes on the default dispatchersaveInCache(news)}// flowOn affects the upstream flow ↑.flowOn(defaultDispatcher)// the downstream flow ↓ is not affected.catch { exception - // Executes in the consumers contextemit(lastCachedNews())}
}七 Jetpack 库中的数据流
Flow with Room 接收有关数据库更改的通知
Dao
abstract class ExampleDao {Query(SELECT * FROM Example)abstract fun getExamples(): FlowListExample
}八 将基于回调的 API 转换为数据流
callbackFlow 是一个数据流构建器允许您将基于回调的 API 转换为数据流。
class FirestoreUserEventsDataSource(private val firestore: FirebaseFirestore
) {// Method to get user events from the Firestore databasefun getUserEvents(): FlowUserEvents callbackFlow {// Reference to use in Firestorevar eventsCollection: CollectionReference? nulltry {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 resumeclose(e)}// Registers callback to firestore, which will be called on new eventsval subscription eventsCollection?.addSnapshotListener { snapshot, _ -if (snapshot null) { returnaddSnapshotListener }// Sends events to the flow! Consumers will get the new eventstry {offer(snapshot.getEvents())} catch (e: Throwable) {// Event couldnt 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 FirestoreawaitClose { subscription?.remove() }}
}
