Kotlin 协程Flow 操作符 ② ( 末端操作符 | collect 操作符 | reduce 操作符 | single 操作符 | first 操作符 | toList 操作符 )

Posted 2022-12-28 韩曙亮

文章目录

• 一、末端操作符
• • 1、collect 操作符
  • 2、reduce 操作符
  • 3、single 操作符
  • 4、first 操作符
  • 5、fold 操作符
  • 6、toList 操作符
  • 7、toSet 操作符

一、末端操作符

---

末端操作符 指的是 在 Flow 流最末端 调用 挂起函数 收集元素 的操作符 , 最常见的 末端操作符 就是 collect 操作符 ;

常见的末端操作符 :

• 收集元素 : collect ;
• 将收集的元素转为集合 : toList , toSet ;
• 收集第一个元素 : first ;
• 发射单个元素 : single ;
• 规约流到单个值 : reduce , fold ;

1、collect 操作符

collect 操作符原型 :

/**
 * 终端流操作符，使用提供的[动作]收集给定的流。
 * 如果在收集过程中或在所提供的流中发生任何异常，则此方法将重新抛出此异常。
 *
 * 使用示例:
 *
 * ```
 * val flow = getMyEvents()
 * try 
 *     flow.collect  value ->
 *         println("Received $value")
 *     
 *     println("My events are consumed successfully")
 *  catch (e: Throwable) 
 *     println("Exception from the flow: $e")
 * 
 * ```
 */
public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =
    collect(object : FlowCollector<T> 
        override suspend fun emit(value: T) = action(value)
    )

2、reduce 操作符

reduce 操作符原型 :

/**
 * 从第一个元素开始累加值，并对当前累加器值和每个元素应用[操作]。
 * 如果流为空，抛出[NoSuchElementException]。
 */
public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S, value: T) -> S): S 
    var accumulator: Any? = NULL

    collect  value ->
        accumulator = if (accumulator !== NULL) 
            @Suppress("UNCHECKED_CAST")
            operation(accumulator as S, value)
         else 
            value
        
    

    if (accumulator === NULL) throw NoSuchElementException("Empty flow can't be reduced")
    @Suppress("UNCHECKED_CAST")
    return accumulator as S

代码示例 :

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)

        runBlocking 
            val result = (0..3)
                        .asFlow()
                        .reducea, b ->
                            a + b
                        
            println("result : $result")
        
    

执行结果 :

2022-12-26 15:51:20.411 24364-24364/kim.hsl.coroutine I/System.out: result : 6

3、single 操作符

single 操作符原型 :

/**
 * 终端操作符，等待一个且仅等待一个值发出。
 * 为空流抛出[NoSuchElementException]，为流抛出[IllegalStateException]
 * 包含多个元素的。
 */
public suspend fun <T> Flow<T>.single(): T 
    var result: Any? = NULL
    collect  value ->
        require(result === NULL)  "Flow has more than one element" 
        result = value
    

    if (result === NULL) throw NoSuchElementException("Flow is empty")
    return result as T

4、first 操作符

first 操作符原型 :

/**
 * 终端操作符，返回流发出的第一个元素，然后取消流的收集。
 * 如果流为空，则抛出[NoSuchElementException]。
 */
public suspend fun <T> Flow<T>.first(): T 
    var result: Any? = NULL
    collectWhile 
        result = it
        false
    
    if (result === NULL) throw NoSuchElementException("Expected at least one element")
    return result as T

5、fold 操作符

fold 操作符原型 :

/**
 * 从[initial]值开始累加值，并应用[operation]当前累加器值和每个元素
 */
public suspend inline fun <T, R> Flow<T>.fold(
    initial: R,
    crossinline operation: suspend (acc: R, value: T) -> R
): R 
    var accumulator = initial
    collect  value ->
        accumulator = operation(accumulator, value)
    
    return accumulator

6、toList 操作符

toList 操作符原型 :

/**
 * 将给定的流收集到[destination]
 */
public suspend fun <T> Flow<T>.toList(destination: MutableList<T> = ArrayList()): List<T> = toCollection(destination)

7、toSet 操作符

toSet 操作符原型 :

/**
 * 将给定的流收集到[destination]
 */
public suspend fun <T> Flow<T>.toSet(destination: MutableSet<T> = LinkedHashSet()): Set<T> = toCollection(destination)