Android Kotlin Flow 如何使用callbackflow

转载请标明出处：http://blog.csdn.net/zhaoyanjun6/article/details/121840157
本文出自【赵彦军的博客】

callbackFlow 原理

callbackFlow：底层使用channel来进行中转，首先通过produce创建一个ReceiveChannel。然后在调用collect的时候，在将channel的值取出来emit出去。

首先我们来看一个例子：

我们有一个网络接口，在耗时3秒后会回调一个结果，把结果 3 ，回调出去。代码如下：

    /**
     * 模拟网络请求
     */
    fun requestApi(block: (Int) -> Unit) {
        thread {
            Thread.sleep(3000)
            block(3)
        }
    }

如果我们要把这个用 Flow 的形式把结果发送出去改怎么写：

        GlobalScope.launch {

            val flow = flow {
                //模拟网络请求
                requestApi {
                    emit(1)
                }
            }

            flow.collect {
                Log.d("list-", "$it")
            }
        }

你以为没问题，其实根本行不通，会报错

报错的意思是：挂起函数只能在协程体内使用，也就是 emit 方法时一个挂起函数，只能用在协程体内。

那该怎么处理呢？用 callbackFlow 就可以了。

什么是 callbackFlow？官方的答案是：将基于回调的 API 转换为数据流。

callbackFlow 是冷流，没有接收者，不会产生数据。

地址是：https://developer.android.com/kotlin/flow

我们来试试看：

        val flow = callbackFlow {

            //模拟网络请求回调
            requestApi { result ->
                //发送数据
                offer(result)
            }
        }

        GlobalScope.launch {
            flow.collect {
                //接收结果 
            }
        }

发现运行起来就崩溃了，抛出以下日志：

java.lang.IllegalStateException: 'awaitClose { yourCallbackOrListener.cancel() }' should be used in the end of callbackFlow block.

意思是：要在 callbackFlow block 体中，调用 awaitClose 方法。

awaitClose 方法是：当数据接收者所在的协程被关闭的时候会调用，作用是：用来释放资源，比如取消网络请求、断开io流、移除监听器、释放内存 等等。

也就是调用 job.cancel() 的时候，会调用 awaitClose。

完整示例如下：

模拟网络请求、模拟取消网络请求

   /**
     * 模拟网络请求
     */
    fun requestApi(block: (Int) -> Unit) {
        thread {
            Log.d("list-", "网络请求")
            Thread.sleep(3000)
            block(3)
        }
    }


  /**
     * 模拟取消网络请求
     */
    fun cancelApi(){
        //do some thing
    }

callbackFlow 如下：

        val flow = callbackFlow {

            //模拟网络请求回调
            requestApi { result ->
                //发送数据
                offer(result)
            }
            
            awaitClose { 
                //当数据接收者所在的协程被关闭的时候会调用。
                //作用是：用来释放资源
                cancelApi()
            }
        }

   val job = GlobalScope.launch {
            flow.collect {
                //接收结果
            }
        }

在产生数据的时候，有两个方法给我用：send 、offer

send ： 必须在协程体内使用offer ： 可以在非协程体内使用

事实上，offer 是 Channel 的方法。

关于 awaitClose

其实很简单嘛，其实就是一个挂起函数，直到接收到 channel 的invokeOnClose回调，才会resume回去。

那具体awaitClose什么时候被回调呢？换句话说，通过 callbackFlow 创建的 flow 什么时候可以被关闭呢？

取消流收集 cancel() 或基于回调的 API 手动调用 SendChannel.close() 时调用或外部的协程被取消时，才会调用awaitClose。换句话说，需要手动关闭创建的callbackFlow，否则就会一直处于运行状态不会结束。

放一个官方的例子：