一、前言

数据流以协程为基础构建，可提供多个值。概念上来讲，数据流可以通过异步方式（挂起函数中执行）进行计算处理一组数据系列（这个与协程挂起函数返回单个值相反），但要求所发出值的类型必须相同。例如，Flow是发出整数值的数据流。

数据流包含是三个实体：

• 提供方：会生产添加到数据流中的数据。得益于协程，数据流还可以异步生成数据。
• 中介（可选）：可以修改发送到数据流的值，或者修正数据流本身
• 使用方：使用数据流中的值

二、数据流使用

2.1 创建数据流

创建数据流，使用数据流构建器对应的API,目前有以下几种：flowOf(…)、asFlow()、flow{…}、channelFlow{…}、MutableStateFlow 和 MutableSharedFlow ，下面我们使用flow{…}来创建数据流，结合官方的例子进行数据流的使用说明.

**需求：**数据源以固定的时间间隔自动获取最新资讯。由于挂起函数只能返回单个值，不能返回多个连续值，因此可以使用数据流来实现该需求。

（1）定义获取网络数据数据接口：

interface NewsApi {

    @GET("/chengyu/query")
    suspend fun fetchLatestNews(
        @Query("key") key: String,
        @Query("word") word: String,
        @Query("dtype") dtype: String
    ): ApiRsp<ArticleHeadline>
}

（2）在Repository中使用定义Flow数据流：

class NewsRepository(
    private val api: NewsApi,
    private val refreshIntervalMs: Long = 5000
) {

    /**
     * 获取数据
     */
    fun loadNews(key: String, word: String, dtype: String): Flow<ApiRsp<ArticleHeadline>> {
        return flow {
            while (true) {
                val latestNews = api.fetchLatestNews(key, word, dtype)
                println("Req ThreadName:"+Thread.currentThread().name)
                emit(latestNews)
                delay(refreshIntervalMs)
            }
        }
    }
}

flow构建器在协程内执行。因此它将受益于相同的异步API，但也存在一些限制：

• 数据流是有序的。当协程内的提供方调用挂起函数时，提供方会挂起，知道挂起函数返回。此示例中，提供方会挂起，直到fetchLatestNews网络请求完成。只有这样，请求结果才能发送到数据流中。
• 使用flow构建器时，提供方不能提供来自不同CoroutineContext的emit值。因此请勿通过创建协程或者使用withContext代码块，在不同的CoroutineContext中调用emit。

2.2 从数据流中进行数据收集

要获取数据流中的值，使用collect。下面展示如何在ViewModel中获取数据

class NewsViewModel() : ViewModel() {
   private val repository = NewsRepository(RetrofitManager.getNewsApi())
    val rspNewsData = MutableLiveData<List<ArticleHeadline>>()

    //加载数据
    fun loadNews() {
        viewModelScope.launch {
            repository.loadNews("7b1aeba080138478034ac887d0aabd9e", "无中生有", "json").collect {
                println("Rsp ThreadName:"+Thread.currentThread().name)
                rspNewsData.value = it
            }
        }
    }

}

在Activity/Fragment中使用数据

class NewsActivity : AppCompatActivity() {
    private val newsViewModel by viewModels<NewsViewModel>()
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_news)
        initObserver()
    }
    fun onClick(view: View) {
        newsViewModel.loadNews()
    }
    private fun initObserver() {
        newsViewModel.rspNewsData.observe(this) {
            println("data is :" + it.result.toString())
        }
    }

}

这里的数据只是将他打印到控制台，运行程序，输出日志如下：

10-12 11:18:59.372 2938-2938/com.example.flowlearn I/System.out: Req ThreadName:main
10-12 11:18:59.373 2938-2938/com.example.flowlearn I/System.out: Rsp ThreadName:main
10-12 11:18:59.374 2938-2938/com.example.flowlearn I/System.out: data is :ArticleHeadline(bushou=一, head=无, pinyin=wú zhōng shēng yǒu, chengyujs= 道家认为，天下万物生于有，有生于无。把没有的说成有。比喻毫无事实，凭空捏造。, from_= 《老子》：“天下万物生于有，有生于无。”, example= 子息从来天数，原非人力能为。最是～，堪令耳目新奇。 明·凌濛初《初刻拍案惊奇》卷三十八)

这里出现一个问题，网络请求的执行居然在main线程，但是程序却没有报错，以前在main线程中执行网络请求，程序直接闪退，现在能正常执行，这里没有搞明白其中的逻辑，但是很明确的是网络请求要放在I/O线程中，那么在数据流中该怎么做呢？

在数据流中要修改CoroutineContext需要使用运算符flowOn。flowOn会更改上游数据的CoroutineContext,表示flowOn之前（或之上）的数据提供方以及其他的中间运算符使用flowOn中的CoroutineContext。下游数据流（晚于flowOn的中间运算符和使用方）则不会收到影响。为了更好的理解这段话，请看下面的图，能够更清晰的表明各段代码片段的执行线程：

在上面的示例中，要要将网络请求切换到IO线程中，我们可以按一下方式添加一个flowOn运算符。

    /**
     * 获取数据
     */
    fun loadNews(key: String, word: String, dtype: String): Flow<ApiRsp<ArticleHeadline>> {
        return flow {
            while (true) {
                val latestNews = api.fetchLatestNews(key, word, dtype)
                println("Req ThreadName:"+Thread.currentThread().name)
                emit(latestNews)
                delay(refreshIntervalMs)
            }
        }.flowOn(Dispatchers.IO)
    }

在运行我们的程序：

10-12 11:52:37.049 3157-3204/com.example.flowlearn I/System.out: Req ThreadName:DefaultDispatcher-worker-2
10-12 11:52:37.052 3157-3157/com.example.flowlearn I/System.out: Rsp ThreadName:main
10-12 11:52:37.052 3157-3157/com.example.flowlearn I/System.out: data is :ArticleHeadline(bushou=一, head=无, pinyin=wú zhōng shēng yǒu, chengyujs= 道家认为，天下万物生于有，有生于无。把没有的说成有。比喻毫无事实，凭空捏造。, from_= 《老子》：“天下万物生于有，有生于无。”, example= 子息从来天数，原非人力能为。最是～，堪令耳目新奇。 明·凌濛初《初刻拍案惊奇》卷三十八)

从上门的日志可以看出，网络请求的操作已经切换到协程分配的DefaultDispatcher-worker-2，而不在main线程中。

2.3 修改数据流

在数据流中有一层叫中介，是个可选项，它是作用是修改数据流。要修改数据流，则需要使用提供的中间运算符函数来执行，比如：map(),zip(),fileter()等，下面通过使用map修改数据流来展示它的应用。

    fun loadNews() {
        viewModelScope.launch {
            repository.loadNews("7b1aeba080138478034ac887d0aabd9e", "无中生有", "json")
                .map {
                    it.reason = "this data changed by map"
                    it
                }
                .collect {
                    println("Rsp ThreadName:" + Thread.currentThread().name)
                    rspNewsData.value = it
                }
        }
    }

运行程序：

10-12 13:48:02.834 3925-4011/com.example.flowlearn I/System.out: Req ThreadName:DefaultDispatcher-worker-1
10-12 13:48:02.835 3925-3925/com.example.flowlearn I/System.out: Rsp ThreadName:main
10-12 13:48:02.835 3925-3925/com.example.flowlearn I/System.out: data is :this data changed by map

通过日志可以看出ApiResponse的数据修改成功。

2.4 数据流异常处理

异常时我们在开发程序中经常遇到的，在协程中不同调用方法，处理异常略有不同，有他们各自的处理方式。在数据流，异常也有它自己的处理方式，使用catch 中间运算符。

下面演示一个不处理异常的情况：

    fun loadNews() {
        viewModelScope.launch {
            repository.loadNews("7b1aeba080138478034ac887d0aabd9e", "无中生有", "json")
                .map {
                    throw Exception("this is custom exception")
                    it
                }
                .collect {
                    println("Rsp ThreadName:" + Thread.currentThread().name)
                    rspNewsData.value = it
                }
        }
    }

在上面的代码中，map 里面抛出一个异常，运行我们的程序，程序会奔溃，会抛出对应的异常信息。

要处理异常我们可以直接try{}catch(){}我们的代码块，比如处理网络请求的异常：

    fun loadNews(key: String, word: String, dtype: String): Flow<ApiRsp<ArticleHeadline>> {
        return flow {
            while (true) {
                try {
                    val latestNews = api.fetchLatestNews(key, word, dtype)
                    println("Req ThreadName:" + Thread.currentThread().name)
                    emit(latestNews)
                    delay(refreshIntervalMs)
                } catch (ex: Throwable) {
                    //handle Exception
                    val apiRsp = ApiRsp<ArticleHeadline>()
                    apiRsp.reason = ex.message
                    emit(apiRsp)
                }

            }
        }.flowOn(Dispatchers.IO)
    }

在上面的网络请求中，针对网络中的各种异常，我们可以封装一个方法处理，给用户友好的提示。

但是当我们数据流中调用一些第三方方法的时候，不可能每个方法都try catch,我们可以使用 catch 中间运算符统一处理所有未捕获的异常。如下：

    fun loadNews() {
        viewModelScope.launch {
            repository.loadNews("7b1aeba080138478034ac887d0aabd9e", "无中生有", "json")
                .map {
                    throw Exception("this is custom exception")
                    it
                }
                .catch { exception ->
                    //处理数据流中未捕获的异常
                    val apiRsp = ApiRsp<ArticleHeadline>()
                    apiRsp.reason = exception.message
                    emit(apiRsp)
                }
                .collect {
                    println("Rsp ThreadName:" + Thread.currentThread().name)
                    rspNewsData.value = it
                }
        }
    }

如果在数据流中catch 运算符捕获了异常，程序不会调用collect的lambda参数，所以需要在catch{}代码块中重新发送数据，这样才会执行collect的lanbda参数。

三、小结

数据流的基本使用如上面说描述的内容，总结起来就以下几点：

1. 如何创建数据流，需要了解不同构建器之间的区别。
2. 如何使用数据流以及结合flowOn修改上游CroutineContext。
3. 如何使用中间运算符修改中介层数据。
4. 数据流的异常处理。

参考资料：

https://developer.android.google.cn/kotlin/flow

最后

以上就是狂野夕阳为你收集整理的数据流-使用入门一、前言二、数据流使用三、小结的全部内容，希望文章能够帮你解决数据流-使用入门一、前言二、数据流使用三、小结所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。