您现在的位置是：首页 > 后端

当前栏目

【Kotlin 协程】Flow 异步流 ⑥ ( 调用 Flow#launchIn 函数指定流收集协程 | 通过取消流收集所在的协程取消流 )

Kotlin 异步函数通过调用指定收集取消

2023-09-14 09:07:26 时间

文章目录

一、调用 Flow#launchIn 函数指定流收集协程
二、通过取消流收集所在的协程取消流

一、调用 Flow#launchIn 函数指定流收集协程

1、指定流收集协程

响应式编程 , 是基于事件驱动的 , 在 Flow 流中会产生源源不断的事件 , 就是发射元素操作 ;

拿到 Flow 流后 , 开始收集元素 , 按照顺序逐个处理产生的事件 ( 元素 ) ;

调用 Flow#launchIn 函数 , 传入协程作用域作为参数 , 可以指定收集 Flow 流元素的协程 ;

在上一篇博客【Kotlin 协程】Flow 异步流 ⑤ 中 , 调用 Flow#flowOn 函数 , 可以指定 Flow 流发射元素的协程 ;

Flow#launchIn 函数返回值是 Job 对象 , 是协程任务对象 , 可调用 Job#cancel 函数取消该协程任务 ;

2、Flow#launchIn 函数原型

Flow#launchIn 函数原型 :

/**
 * 终端流操作符，在[作用域]中[启动][启动]给定流的[收集][收集]。
 * 它是“范围”(scope)的简称。启动{flow.collect()} '。
 *
 * 此操作符通常与[onEach]， [onCompletion]和[catch]操作符一起使用，以处理所有发出的值
 * 处理上游流或处理过程中可能发生的异常，例如:
 *
 * ```
 * flow
 *     .onEach { value -> updateUi(value) }
 *     .onCompletion { cause -> updateUi(if (cause == null) "Done" else "Failed") }
 *     .catch { cause -> LOG.error("Exception: $cause") }
 *     .launchIn(uiScope)
 * ```
 *
 * 注意，[launchIn]的结果值没有被使用，提供的作用域负责取消。
 */
public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = scope.launch {
    collect() // tail-call
}

3、代码示例

代码示例 :

package kim.hsl.coroutine

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        // 携程中调用挂起函数返回一个 Flow 异步流
        runBlocking {
            println("流收集时的协程上下文 : ${Thread.currentThread().name}")

            flowEvent()
                .onEach {
                    // 逐个处理产生的事件
                    println("接收到事件 : $it, 当前线程 : ${Thread.currentThread().name}")
                }
                .launchIn(CoroutineScope(Dispatchers.IO)) // 在指定的协程作用域中处理收集元素操作,
                                                          // 该 launchIn 函数返回一个 Job 对象
                .join()     // 该协程不是 runBlocking 主协程 的子协程, 需要调用 join 等待协程执行完毕
        }
    }

    /**
     * 使用 flow 构建器 Flow 异步流
     * 产生事件的 事件源
     */
    suspend fun flowEvent() = (0..3)
        .asFlow()                               // 将区间转为 Flow 流
        .onEach {
            delay(500)
            println("发射事件 : $it, 当前线程 : ${Thread.currentThread().name}")
        }      // 发射元素 ( 产生事件 ) 时挂起 500ms
        .flowOn(Dispatchers.Default)           // 设置发射元素的协程
}

执行结果 :

2022-12-23 16:06:58.720 2950-2950/kim.hsl.coroutine I/System.out: 流收集时的协程上下文 : main
2022-12-23 16:06:59.345 2950-3080/kim.hsl.coroutine I/System.out: 发射事件 : 0, 当前线程 : DefaultDispatcher-worker-3
2022-12-23 16:06:59.347 2950-3078/kim.hsl.coroutine I/System.out: 接收到事件 : 0, 当前线程 : DefaultDispatcher-worker-1
2022-12-23 16:06:59.885 2950-3078/kim.hsl.coroutine I/System.out: 发射事件 : 1, 当前线程 : DefaultDispatcher-worker-1
2022-12-23 16:06:59.887 2950-3079/kim.hsl.coroutine I/System.out: 接收到事件 : 1, 当前线程 : DefaultDispatcher-worker-2
2022-12-23 16:07:00.394 2950-3080/kim.hsl.coroutine I/System.out: 发射事件 : 2, 当前线程 : DefaultDispatcher-worker-3
2022-12-23 16:07:00.396 2950-3080/kim.hsl.coroutine I/System.out: 接收到事件 : 2, 当前线程 : DefaultDispatcher-worker-3
2022-12-23 16:07:00.938 2950-3078/kim.hsl.coroutine I/System.out: 发射事件 : 3, 当前线程 : DefaultDispatcher-worker-1
2022-12-23 16:07:00.940 2950-3079/kim.hsl.coroutine I/System.out: 接收到事件 : 3, 当前线程 : DefaultDispatcher-worker-2

在这里插入图片描述

二、通过取消流收集所在的协程取消流

Flow 流的收集元素操作 , 是在协程中执行 , 将协程取消 , 即可将 Flow 流收集操作取消 , 也就是将 Flow 流取消 ;

代码示例 : 使用 withTimeoutOrNull(2000) 创建一个协程 , 该协程在 2000ms 后自动超时取消 , 同时在其中进行流收集的操作也一并取消 ;

package kim.hsl.coroutine

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        // 携程中调用挂起函数返回一个 Flow 异步流
        runBlocking {
            // 该协程作用域 2 秒后超时取消
            withTimeoutOrNull(2000){
                flowEvent().collect {
                    println("收集到元素 : $it")
                }
            }
            println("协程作用域取消")
        }
    }

    /**
     * 使用 flow 构建器 Flow 异步流
     * 产生事件的 事件源
     */
    suspend fun flowEvent() = (0..10)
        .asFlow()                               // 将区间转为 Flow 流
        .onEach {
            delay(500)
            println("发射事件 : $it, 当前线程 : ${Thread.currentThread().name}")
        }      // 发射元素 ( 产生事件 ) 时挂起 500ms
        .flowOn(Dispatchers.Default)           // 设置发射元素的协程
}