Flow创建方式

1.1 创建Flow方式1

safeLaunch({
    flow {
        for (i in 1..5) {
            delay(100)
            Log.e(
                TAG,
                "-----------------flow------------${Thread.currentThread().name}-----------"
            )
            emit(i)
        }
    }.collect {
        Log.e(
            TAG,
            "-------collect---------$it-------------${Thread.currentThread().name}-----------"
        )
    }
})

image.png

1.2创建Flow方式2

fun createFlow2() {

    safeLaunch({
        flowOf(1, 2, 3, 4, 5, 6).collect {
            Log.e(
                TAG,
                "----------------$it-------------${Thread.currentThread().name}-----------"
            )
        }
    })

}

1.3创建Flow方式3

fun createFlow3() {
    safeLaunch({
        listOf(1, 2, 3, 4, 5, 6,7).asFlow().collect {
            Log.e(
                TAG,
                "----------------$it-------------${Thread.currentThread().name}-----------"
            )
        }
    })
}

二、切换线程

fun createFlow() {

    safeLaunch({
        flow {
            for (i in 1..5) {
                delay(100)
                Log.e(TAG, "-----------------flow--$i----------${Thread.currentThread().name}-----------")
                emit(i)
                if (i==3){
                    i/0
                }
            }
        }.map{
           val value= it*it
            Log.e(TAG, "-------map1-------$value--------------${Thread.currentThread().name}-----------")
            value
        }.flowOn(Dispatchers.IO).map{
            val value= it-1
            Log.e(TAG, "-------map2-------$value--------------${Thread.currentThread().name}-----------")
            value
        }.onStart {
            Log.e(TAG, "-------onStart---------------------${Thread.currentThread().name}-----------")
        }.catch {
            Log.e(TAG, "-------catch---------$it-------------${Thread.currentThread().name}-----------")
        }.onCompletion {
            Log.e(TAG, "-------onCompletion---------------------${Thread.currentThread().name}-----------")
        }.flowOn(Dispatchers.Main).collect {
            Log.e(TAG, "-------collect---------$it-------------${Thread.currentThread().name}-----------")
        }
    })

}

image.png

注意：

1、flowOn 操作符，接收一个线程调度参数，影响的是上游的操作
2、collect() 指定哪个线程，则需要看整个 flow 处于哪个 CoroutineScope 下。

三、异常捕获

fun createFlow() {

    safeLaunch({
        flow {
            for (i in 1..5) {
                delay(100)
                Log.e(TAG, "-----------------flow------------${Thread.currentThread().name}-----------")
                emit(i)
                if (i==5){
                    i/0
                }
            }
        }.catch {
            Log.e(TAG, "-------catch---------$it-------------${Thread.currentThread().name}-----------")
        }.flowOn(Dispatchers.IO).collect {
            Log.e(TAG, "-------collect---------$it-------------${Thread.currentThread().name}-----------")
        }
    })

}

image.png

四、Flow生命周期监听

4.1 发生异常后 map2 collect 发生几率不确定

    fun createFlow() {

        safeLaunch({
            flow {
                for (i in 1..5) {
                    delay(100)
                    Log.e(TAG, "-----------------flow--$i----------${Thread.currentThread().name}-----------")
                    emit(i)
                    if (i==3){
                        i/0
                    }
                }
            }.map{
               val value= it
                Log.e(TAG, "-------map1-------$value--------------${Thread.currentThread().name}-----------")
                value
            }.flowOn(Dispatchers.IO).map{
                val value= it
                Log.e(TAG, "-------map2-------$value--------------${Thread.currentThread().name}-----------")
                value
            }.onStart {
                Log.e(TAG, "-------onStart---------------------${Thread.currentThread().name}-----------")
            }.catch {
                Log.e(TAG, "-------catch---------$it-------------${Thread.currentThread().name}-----------")
            }.onCompletion {
                Log.e(TAG, "-------onCompletion---------------------${Thread.currentThread().name}-----------")
            }.flowOn(Dispatchers.Main).collect {
                Log.e(TAG, "-------collect---------$it-------------${Thread.currentThread().name}-----------")
            }
        })

    }

image.png
image.png

4.1 发生异常后 map2 collect 发生确定

    fun createFlow() {

        safeLaunch({
            flow {
                for (i in 1..5) {
                    delay(100)
                    Log.e(TAG, "-----------------flow--$i----------${Thread.currentThread().name}-----------")
                    emit(i)
                    if (i==3){
                        i/0
                    }
                }
            }.map{
               val value= it
                Log.e(TAG, "-------map1-------$value--------------${Thread.currentThread().name}-----------")
                value
            }.catch {
                Log.e(TAG, "-------catch---------$it-------------${Thread.currentThread().name}-----------")
            }.flowOn(Dispatchers.IO).map{
                val value= it
                Log.e(TAG, "-------map2-------$value--------------${Thread.currentThread().name}-----------")
                value
            }.onStart {
                Log.e(TAG, "-------onStart---------------------${Thread.currentThread().name}-----------")
            }.onCompletion {
                Log.e(TAG, "-------onCompletion---------------------${Thread.currentThread().name}-----------")
            }.flowOn(Dispatchers.Main).collect {
                Log.e(TAG, "-------collect---------$it-------------${Thread.currentThread().name}-----------")
            }
        })

    }

image.png