JetPack知识点实战系列十二：Kotlin Flow基础知识详解

69 阅读 0 评论 46 点赞

我是靠谱客的博主缥缈方盒，最近开发中收集的这篇文章主要介绍JetPack知识点实战系列十二：Kotlin Flow基础知识详解，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

阅读过实战系列三的文章的同学可能有印象，我有提到过Kotlin Flow和Room数据库能很好的配合使用。此外，不久前官方推出了DataStore也是基于Kotlin Flow，所以本节我们先来了解下Kotlin Flow的使用方法。

Flow 引入的场景

Kotlin的官方解释引入Flow的场景是：

Return multiple asynchronously computed values over time

这句话很重要，它包括几个关键含义：

返回值有多个
返回值是分开的
返回的是异步结果

接下来我们看看单个含义在Kotlin中如何实现：

返回多个值 - Collections

返回多个值我们能很容易的想到使用Collections。例如一个
返回List的listExample的函数。

fun listExample(): List<Int> {
    return listOf(1, 2, 3)
}

listExample().forEach { value -> println(value) }
// 1, 2, 3

值是分开返回的

上面的情况多个值是作为一个整体一起返回的，如果让多个值分开返回就得使用Kotlin的另外一个数据类型Sequences。

fun sequenceExample(): Sequence<Int> {
    return sequence {
        Thread.sleep(1000)  // 假装在执行耗时操作
        yield(1)            // 返回值
        Thread.sleep(1000)
        yield(2)
        Thread.sleep(1000)
        yield(3)
    }
}

sequenceExample().forEach { value ->
    println(value)
    println(Date().time)
}

// 结果：
I/System.out: 1
I/System.out: 1605153441181
I/System.out: 2
I/System.out: 1605153442182
I/System.out: 3
I/System.out: 1605153443185

结果可以看出，通过使用sequence可以实现多个结果分批次单个数值的返回。

异步返回

异步返回可以使用susupend函数。

suspend fun simpleAsyn(): List<Int> {
    delay(1000)     // 假装在执行异步耗时操作
    return listOf(1, 2, 3)
}

Flow实现Return multiple asynchronously computed values over time

fun simpleFlow(): Flow<Int> {
    return flow {
        delay(1000) // 假装在执行异步耗时操作
        emit(1)     // 返回值
        delay(1000)
        emit(2)
        delay(1000)
        emit(3)
    }
}
    
simpleFlow().collect { value -> println(value) }    
    
// 1, 2, 3

结论：Flow可以简单的理解为异步的Sequence。

Flow 的特性

Flow可以理解成异步的Sequence，也具有和Sequence类似的一些特性。

延迟执行

我们的代码如下，打印结果出来可能会比较费解。

val streamOfInts: Flow<Int> = flow {
    println("开始发送===>")
    for (i in 1..3) {
        delay(100)
        println("发送初始值===>$i")
        emit(i)
        }
        println("结束发送===>")
    }
    
streamOfInts.map { value -> value * value }

println("Flow完成===>")

// 打印结果:
println("Flow完成===>")

为什么呢? 因为Flow是惰性的，streamOfInts这个Flow没有执行结束操作函数，Flow中的各种中间操作函数不会立即执行

结束操作函数 — Flow执行函数后的返回结果不是Flow，那这个函数就是结束操作函数

中间操作函数 — Flow执行函数后的返回结果仍然是Flow

我们接下来修改代码，执行collect这个函数，它的意义是接收Flow传过来的值：

val streamOfInts: Flow<Int> = flow {
    println("开始发送===>")
    for (i in 1..3) {
        delay(100)
        println("发送初始值===>$i")
        emit(i)
        }
        println("结束发送===>")
    }
    
streamOfInts.map { value -> value * value }.collect { println("最终收到 ===> $it") }

// 打印结果:
I/System.out: 开始发送===>
I/System.out: 发送初始值===>1
I/System.out: 最终收到 ===> 1
I/System.out: 发送初始值===>2
I/System.out: 最终收到 ===> 4
I/System.out: 发送初始值===>3
I/System.out: 最终收到 ===> 9
I/System.out: 结束发送===>
I/System.out: Flow完成===>

熟悉RxJava的同学是不是似曾相识的感觉，没错Flow就是冷信号，只有被订阅后才会发送值。

顺序执行

上面的结果我们能看出来，分别对每个值进行了所有的操作后才进行下一个值的处理，即先处理完1，再处理2，再处理3。

1 - 1 - 2 - 4 - 3 - 9

Flow的构造函数

通过flowOf函数

flowOf(1, 2, 3)
// 1, 2, 3
flowOf(listOf(1,2,3))
// [1, 2, 3]

Iterable调用asFlow函数

listOf(1, 2, 3).asFlow()
// 1, 2, 3

无参但是有返回值的函数**(() -> T)**调用asFlow函数

fun flowBuilderFunction(): Int {
    return 10
}

::flowBuilderFunction.asFlow()

// 10

无参但是有返回值的挂起函数**(() -> T)**调用asFlow函数

suspend fun flowBuilderFunction(): Int {
    return 10
}

::flowBuilderFunction.asFlow()

// 10

Array调用asFlow函数

arrayOf(1,2,3).asFlow()
// 1, 2, 3

LongRange调用asFlow函数

LongRange(1, 5).asFlow().collect { value -> println(value) }

//  1, 2, 3, 4, 5

Empty Flow

emptyFlow<String>()

Flow中间运算函数

Delay相关的运算函数

debounce

如果两个相邻的值生产出来的时间间隔超过了[timeout]毫秒，就忽过滤掉前一个值
最后一个值不受影响，总是会被释放emit。

flow {
    emit(1)
    delay(3000)
    emit(2)
    delay(1000)
    emit(3)
    delay(1000)
    emit(4)
}.debounce(2000)

// 结果：1 4 
// 解释：
//  2和1的间隔大于2000，1被释放
//  3和2的间隔小于2000, 2被忽略
//  4和3的间隔小于2000, 3被忽略
//  4是最后一个值不受timeout值的影响, 4被释放

[timeout]可以传毫秒，也可以传Duration

注意：还是个实验性的方法,使用的时候需要加上@ExperimentalTime

flow {
    emit(1)
    delay(3000)
    emit(2)
    delay(1000)
    emit(3)
    delay(1000)
    emit(4)
}.debounce(2000.milliseconds)

// 结果：1 4

Distinct相关的运算函数

distinctUntilChanged

如果生产的值和上个发送的值相同，值就会被过滤掉

flow {
    emit(1)
    emit(1)
    emit(2)
    emit(2)
    emit(3)
    emit(4)
}.distinctUntilChanged()

// 结果：1 2 3 4
// 解释：
// 第一个1被释放
// 第二个1由于和第一个1相同，被过滤掉
// 第一个2被释放
// 第二个2由于和第一个2相同，被过滤掉
// 第一个3被释放
// 第一个4被释放

可以传参(old: T, new: T) -> Boolean，进行自定义的比较

private class Person(val age: Int, val name: String)

flow {
    emit(Person(20, "张三"))
    emit(Person(21, "李四"))
    emit(Person(21, "王五"))
    emit(Person(22, "赵六"))
}.distinctUntilChanged{old, new -> old.age == new.age }
.collect{ value -> println(value.name) }
    
// 结果：张三 李四 赵六
// 解释：本例子定义如果年龄相同就认为是相同的值，所以王五被过滤掉了

可以用distinctUntilChangedBy转换成年龄进行对比

flow {
    emit(Person(20, "张三"))
    emit(Person(21, "李四"))
    emit(Person(21, "王五"))
    emit(Person(22, "赵六"))
}.distinctUntilChangedBy { person -> person.age }

// 结果：张三 李四 赵六

Emitters相关的运算函数

transform

对每个值进行转换

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.transform {
    if (it % 2 == 0) {
        emit(it * it)
    }
}

// 结果：4 16
// 解释：
// 1 不是偶数，被忽略
// 2 是偶数，2的平方4
// 3 不是偶数，被忽略
// 4 是偶数，4的平方16

onStart

第一个值被释放之前被执行

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.onStart { emit(1000) }

// 结果：1000 1 2 3 4
// 解释：
// 第一个值1被释放的时候调用了emit(1000), 所以1000在1之前被释放

onCompletion

最后一个值释放完成之后被执行

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.onCompletion { emit(1000) }

// 结果：1 2 3 4 1000
// 解释：
// 第一个值4被释放的时候调用了emit(1000), 所以1000在4之后被释放

Limit相关的运算函数

drop

忽略最开始的[count]个值

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.drop(2)

// 结果：3 4
// 解释：
// 最开始释放的两个值(1,2)被忽略了

dropWhile

判断第一个值如果满足(T) -> Boolean这个条件就忽略

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.dropWhile {
    it % 2 == 0
}

// 结果：1 2 3 4
// 解释：
// 第一个值不是偶数，所以1被释放

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.dropWhile {
    it % 2 != 0
}

// 结果：2 3 4
// 解释：
// 第一个值是偶数，所以1被忽略

take

只释放前面[count]个值

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.take(2)

// 结果：1 2
// 解释：
// 前面两个值被释放

takeWhile

判断第一个值如果满足(T) -> Boolean这个条件就释放

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.takeWhile { it%2 != 0 }

// 结果：1
// 解释：
// 第一个值满足是奇数条件

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.takeWhile { it%2 == 0 }

// 结果：无
// 解释：
// 第一个值不满足是奇数条件

CoroutineContext相关的运算函数

flowOn

可以切换CoroutineContext

说明：flowOn只影响该运算符之前的CoroutineContext，对它之后的CoroutineContext没有任何影响

withContext(Dispatchers.Main) {
    flow {
        println("Thread name1 = ${Thread.currentThread().name}")
        emit(1)
        emit(2)
        emit(3)
        emit(4)
    }
        .map {
            println("Thread name2 = ${Thread.currentThread().name}")
            it * it
        }
        .filter {
            println("Thread name3 = ${Thread.currentThread().name}")
            it > 9
        }
        .flowOn(Dispatchers.IO)
        .collect{value ->
            println("Thread name4 = ${Thread.currentThread().name}")
            println(value)
        }
}


// 结果：
// Thread name1 = DefaultDispatcher-worker-1
// Thread name2 = DefaultDispatcher-worker-1
// Thread name3 = DefaultDispatcher-worker-1
// Thread name2 = DefaultDispatcher-worker-1
// Thread name3 = DefaultDispatcher-worker-1
// Thread name2 = DefaultDispatcher-worker-1
// Thread name3 = DefaultDispatcher-worker-1
// Thread name2 = DefaultDispatcher-worker-1
// Thread name3 = DefaultDispatcher-worker-1
// Thread name4 = main
// 16
// 解释：
// flowOn(Dispatchers.IO)之前的flow，map，filter都是在Dispatchers.IO上执行
// flowOn(Dispatchers.IO)之后的collect则由withContext(Dispatchers.Main)决定

buffer

将flow的多个任务分配到不同的协程中去执行，加快执行的速度。

val flow1 = flow {
    delay(2000)  // 假设耗时任务
    emit(1)      // 释放值
    delay(2000) 
    emit(2)
    delay(2000)
    emit(3)
    delay(2000)
    emit(4)
}

flow1.collect { value ->
    println(value)
    delay(4000)
}

// 结果
// 2020-11-16 13:48:37.144 24060-24116/com.johnny.flowdemo I/System.out: 1
// 2020-11-16 13:48:43.150 24060-24116/com.johnny.flowdemo I/System.out: 2
// 2020-11-16 13:48:49.160 24060-24116/com.johnny.flowdemo I/System.out: 3
// 2020-11-16 13:48:55.166 24060-24116/com.johnny.flowdemo I/System.out: 4
// 解释：
// 4个耗时操作每个2000毫秒，加上collect的4000毫秒，所以每个值的时间间隔是6000毫秒。

val flow2 = flow {
    delay(2000)  // 假设耗时任务
    emit(1)      // 释放值
    delay(2000) 
    emit(2)
    delay(2000)
    emit(3)
    delay(2000)
    emit(4)
}.buffer()

flow2.collect { value ->
    println(value)
    delay(4000)
}

// 结果
// 2020-11-16 13:51:11.290 24253-24299/com.johnny.flowdemo I/System.out: 1
// 2020-11-16 13:51:15.293 24253-24299/com.johnny.flowdemo I/System.out: 2
// 2020-11-16 13:51:19.297 24253-24300/com.johnny.flowdemo I/System.out: 3
// 2020-11-16 13:51:23.301 24253-24300/com.johnny.flowdemo I/System.out: 4

// 解释：
// 4个耗时操作被分配到了不同的协程中执行，总共耗时了大约2000毫秒。collect收到的4个值差不多同时，所以每个值依次收到的时间间隔是4000毫秒。

conflate

如果值的生产速度大于值的消耗速度，就忽略掉中间未来得及处理的值，只处理最新的值。

val flow1 = flow {
    delay(2000)
    emit(1)
    delay(2000)
    emit(2)
    delay(2000)
    emit(3)
    delay(2000)
    emit(4)
}.conflate()

flow1.collect { value ->
    println(value)
    delay(5000)
}

// 结果: 1 3 4
// 解释：
// 2000毫秒后生产了1这个值，交由collect执行，花费了5000毫秒，当1这个值执行collect完成后已经经过了7000毫秒。
// 这7000毫秒中，生产了2，但是collect还没执行完成又生产了3，所以7000毫秒以后会直接执行3的collect方法，忽略了2这个值
// collect执行完3后，还有一个4，继续执行。

Flatten相关的运算函数

flatMapConcat

将原始的Flow<T>通过[transform]转换成Flow<Flow<T>>，然后将Flow<Flow<T>>释放的Flow<T>其中释放的值一个个释放。

flow {
    delay(1000)
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(1000)
    emit(4)
}.flatMapConcat {
    flow {
        emit("$it 产生第一个flow值")
        delay(2500)
        emit("$it 产生第二个flow值")
    }
}.collect { value ->
    println(value)
}

// 结果
// I/System.out: 1 产生第一个flow值
// I/System.out: 1 产生第二个flow值
// I/System.out: 2 产生第一个flow值
// I/System.out: 2 产生第二个flow值
// I/System.out: 3 产生第一个flow值
// I/System.out: 3 产生第二个flow值
// I/System.out: 4 产生第一个flow值
// I/System.out: 4 产生第二个flow值

// 解释：
// 原始Flow<Int>通过flatMapConcat被转换成Flow<Flow<Int>>
// 原始Flow<Int>首先释放1，接着Flow<Flow<Int>> 就会释放 1产生第一个flow值 和 1产生第二个flow值 两个值
// Flow<Int>释放2，...
// Flow<Int>释放3，...
// Flow<Int>释放4，...

flattenConcat

和flatMapConcat类似，只是少了一步Map操作。

flow {
    delay(1000)
    emit(flow {
        emit("1 产生第一个flow值")
        delay(2000)
        emit("1 产生第二个flow值") })
    delay(1000)
    emit(flow {
        emit("2 产生第一个flow值")
        delay(2000)
        emit("3 产生第二个flow值") })
    delay(1000)
    emit(flow {
        emit("3 产生第一个flow值")
        delay(2000)
        emit("3 产生第二个flow值") })
    delay(1000)
    emit(flow {
        emit("4 产生第一个flow值")
        delay(2500)
        emit("4 产生第二个flow值") })
    }.flattenConcat()
    
// 结果
// I/System.out: 1 产生第一个flow值
// I/System.out: 1 产生第二个flow值
// I/System.out: 2 产生第一个flow值
// I/System.out: 2 产生第二个flow值
// I/System.out: 3 产生第一个flow值
// I/System.out: 3 产生第二个flow值
// I/System.out: 4 产生第一个flow值
// I/System.out: 4 产生第二个flow值

flatMapMerge

将原始的Flow<T>通过[transform]转换成Flow<Flow<T>>，然后将Flow<Flow<T>>释放的Flow<T>其中释放的值一个个释放。

它与flatMapConcat的区别是：Flow<Flow<T>>释放的Flow<T>其中释放的值没有顺序性，谁先产生谁先释放。

flow {
    delay(1000)
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(1000)
    emit(4)
}.flatMapMerge {
    flow {
        emit("$it 产生第一个flow值")
        delay(2500)
        emit("$it 产生第二个flow值")
    }
}.collect { value ->
    println(value)
}

// 结果
// I/System.out: 1 产生第一个flow值
// I/System.out: 2 产生第一个flow值
// I/System.out: 3 产生第一个flow值
// I/System.out: 1 产生第二个flow值
// I/System.out: 4 产生第一个flow值
// I/System.out: 2 产生第二个flow值
// I/System.out: 3 产生第二个flow值
// I/System.out: 4 产生第二个flow值

// 解释：
// 原始Flow<Int>首先释放1, 第二个Flow<Flow<Int>> 释放 1产生第一个flow值，但是 1产生第二个flow值是3500毫秒才释放，2 产生第一个flow值 是2000毫秒释放， 3 产生第一个flow值 是3000毫秒释放，3500毫秒时刻才是 1产生第二个flow值 的释放

flatMapMerge

和 flattenMerge类似，只是少了一步Map操作。

merge

将Iterable<Flow<T>>合并成一个Flow<T>

val flow1 = listOf(
    flow {
        emit(1)
        delay(500)
        emit(2)
    },
    flow {
        emit(3)
        delay(500)
        emit(4)
    },
    flow {
        emit(5)
        delay(500)
        emit(6)
    }
)
flow1.merge().collect { value -> println("$value") }

// 结果: 1 3 5 2 4 6
// 解释：
// 按Iterable的顺序和耗时顺序依次释放值

transformLatest

原始flow会触发transformLatest转换后的flow, 当原始flow有新的值释放后，transformLatest转换后的flow会被取消，接着触发新的转换后的flow

flow {
    emit(1)
    delay(1000)
    emit(2)
    delay(2000)
    emit(3)
    delay(3000)
    emit(4)
}.transformLatest { value ->
    delay(2500)
    emit(value * value )
}

// 结果: 9 16
// 解释：
// 原始Flow释放1以后，转换后的Flow还没来得及释放1，原始Flow释放2
// 原始Flow释放2以后，转换后的Flow还没来得及释放4，原始Flow释放3
// 原始Flow释放3以后，转换后的Flow有足够的时间释放9
// 原始Flow释放4以后，转换后的Flow有足够的时间释放16

flatMapLatest

和transformLatest类似，原始flow会触发transformLatest转换后的flow, 当原始flow有新的值释放后，transformLatest转换后的flow会被取消，接着触发新的转换后的flow

区别：flatMapLatest的transform转换成的是Flow<T>, transformLatest的transform转换成的是Unit

flow {
    emit(1)
    delay(1000)
    emit(2)
    delay(2000)
    emit(3)
    delay(3000)
    emit(4)
}.flatMapLatest { value ->
    flow {
        delay(2500)
        emit(value * value )
    }
}

// 结果: 9 16
// 解释：
// 原始Flow释放1以后，转换后的Flow还没来得及释放1，原始Flow释放2
// 原始Flow释放2以后，转换后的Flow还没来得及释放4，原始Flow释放3
// 原始Flow释放3以后，转换后的Flow有足够的时间释放9
// 原始Flow释放4以后，转换后的Flow有足够的时间释放16

mapLatest

和transformLatest类似，原始flow会触发transformLatest转换后的flow, 当原始flow有新的值释放后，transformLatest转换后的flow会被取消，接着触发新的转换后的flow

区别：mapLatest的transform转换成的是T,flatMapLatest的transform转换成的是Flow<T>,transformLatest的transform转换成的是Unit

Transform相关的运算函数

filter

通过predicate进行过滤，满足条件则被释放

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.filter { it % 2 == 0 }

// 结果: 2 4
// 解释：
// 2和4满足it % 2 == 0，被释放

filterNot

通过predicate进行过滤，不满足条件则被释放

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.filterNot { it % 2 == 0 }

// 结果: 1 3
// 解释：
// 1和3不满足it % 2 == 0，被释放

filterIsInstance

如果是某个数据类型则被释放

flow {
    emit(1)
    emit("2")
    emit("3")
    emit(4)
}.filterIsInstance<String>()

// 结果: "2" "3"
// 解释：
// "2" "3"是String类型，被释放

filterNotNull

如果数据是非空，则被释放

flow {
    emit(1)
    emit("2")
    emit("3")
    emit(null)
}.filterNotNull()

// 结果: 1 "2" "3"

map

将一个值转换成另外一个值

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.map { it * it }

// 结果: 1 4 9 16
// 解释：
// 将1，2，3，4转换成对应的平方数

mapNotNull

将一个非空值转换成另外一个值

withIndex

将值封装成IndexedValue对象

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.withIndex()

// 结果:
// I/System.out: IndexedValue(index=0, value=1)
// I/System.out: IndexedValue(index=1, value=2)
// I/System.out: IndexedValue(index=2, value=3)
// I/System.out: IndexedValue(index=3, value=4)

onEach

每个值释放的时候可以执行的一段代码

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.onEach { println("接收到$it") }

// 结果:
I/System.out: 接收到1
I/System.out: 1
I/System.out: 接收到2
I/System.out: 2
I/System.out: 接收到3
I/System.out: 3
I/System.out: 接收到4
I/System.out: 4

scan

有一个初始值，然后每个值都和初始值进行运算，然后这个值作为后一个值的初始值

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.scan(100) { acc, value ->
    acc * value
}

// 结果: 100 100 200 600 2400
// 解释：
// 初始值 100
// 1  100 * 1 = 100
// 2  100 * 2 = 200
// 3  200 * 3 = 600
// 4  600 * 4 = 2400

runningReduce

和scan类似，但是没有初始值，最开始是它本身

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.runningReduce { acc, value ->
    acc * value
}

// 结果: 1 2 6 24
// 解释：
// 1  1
// 2  1 * 2 = 2
// 3  2 * 3 = 6
// 4  6 * 4 = 24

Combine相关的运算函数

任意一个flow释放值且都有释放值后会调用combine后的代码块，且值为每个flow的最新值。

val flow1 = flowOf(1, 2, 3, 4).onEach { delay(10) }
val flow2 = flowOf("a", "b", "c", "d").onEach { delay(20) }

flow1.combine(flow2) { first, second ->
    "$first$second"
}.collect { println("$it") }

// 结果：1a 2a 2b 3b 4b 4c 4d

// 解释：
// 开始 --- flow1 释放 1，flow2 释放 a, 释放1a
// 10毫秒 --- flow1 释放 2，释放2a
// 20毫秒 --- flow2 释放 b，此时释放2b
// 30毫秒 --- flow1 释放 3，此时释放3b
// 40毫秒 --- flow1 释放 4，此时释放4b
// 40毫秒 --- flow2 释放 c，此时释放4c
// 60毫秒 --- flow2 释放 d，此时释放4d

说明：Combine能够接受的参数类型非常多，但是效果都是如上类似。

Flow结束函数

Collect相关的结束函数

collect

接收值，一直有用，无需再做介绍。

launchIn

scope.launch { flow.collect() }的缩写, 代表在某个协程上下文环境中去接收释放的值

val flow1 = flow {
    delay(1000)
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(1000)
    emit(4)
}

flow1.onEach { println("$it") }
    .launchIn(GlobalScope)

// 结果：1 2 3 4

collectIndexed

和withIndex对应的，接收封装的IndexedValue

val flow1 = flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}.withIndex()

flow1.collectIndexed { index, value ->
    println("index = $index, value = $value")
}

// 结果：
// I/System.out: index = 0, value = IndexedValue(index=0, value=1)
// I/System.out: index = 1, value = IndexedValue(index=1, value=2)
// I/System.out: index = 2, value = IndexedValue(index=2, value=3)
// I/System.out: index = 3, value = IndexedValue(index=3, value=4)

collectLatest

collectLatest与collect的区别是，如果有新的值释放，上一个值的操作如果没执行完则将会被取消

val flow1 = flow {
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(2000)
    emit(4)
}

flow1.collectLatest {
    println("正在计算收到的值 $it")
    delay(1500)
    println("收到的值 $it")
}

// 结果：
// I/System.out: 正在计算收到的值 1
// I/System.out: 正在计算收到的值 2
// I/System.out: 正在计算收到的值 3
// I/System.out: 收到的值 3
// I/System.out: 正在计算收到的值 4
// I/System.out: 收到的值 4

// 解释：
// 1间隔1000毫秒后释放2，2间隔1000毫秒后释放3，这间隔小于需要接收的时间1500毫秒，所以当2和3 到来后，之前的操作被取消了。
// 3和4 之间的间隔够长能够等待执行完毕，4是最后一个值也能执行

Collection相关的结束函数

toList

将释放的值转换成List

flow {
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(2000)
    emit(4)
}

println(flow1.toList())

// 结果：[1, 2, 3, 4]

toSet

将释放的值转换成Set

flow {
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(2000)
    emit(4)
}

println(flow1.toSet())

// 结果：[1, 2, 3, 4]

Count相关的结束函数

count

计算释放值的个数

val flow1 = flow {
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(2000)
    emit(4)
}

println(flow1.count())
        
// 结果：4

计算满足某一条件的释放值的个数

val flow1 = flow {
    emit(1)
    delay(1000)
    emit(2)
    delay(1000)
    emit(3)
    delay(2000)
    emit(4)
}

println(flow1.count { it % 2 == 0 })
        
// 结果：2
// 解释：
// 偶数有2个值 2 4

Reduce相关的结束函数

reduce

和runningReduce类似，但是只计算最后的结果。

val flow1 = flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}
println(flow1.reduce { acc, value -> acc * value })

// 结果：24
// 解释：计算最后的结果，1 * 2 * 3 * 4 = 24

fold

和scan类似，有一个初始值，但是只计算最后的结果。

val flow1 = flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}
println(flow1.fold(100) { acc, value -> acc * value })

// 结果：2400
// 解释：计算最后的结果，100 * 1 * 2 * 3 * 4 = 2400

single

只接收一个值的Flow

注意：多于1个或者没有值都会报错

 val flow1 = flow {
    emit(1)
}
println(flow1.single())

// 结果：1

singleOrNull

接收一个值的Flow或者一个空值的Flow

first/firstOrNull

接收释放的第一个值/接收第一个值或者空值

val flow1 = flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}
println(flow1.first())

// 结果：1

接收第一个满足某个条件的值

val flow1 = flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
}
println(flow1.first { it % 2 == 0})

// 结果：2

Flow的错误异常处理

可以通过 try catch 捕获错误异常

try {
    flow {
        emit(1)
        emit(2)
        emit(3)
        emit(4)
    }.collect {
        println("接收值 $it")
        check(it <= 1) { "$it 大于1"  }
    }
} catch (e: Throwable) {
    println("收到了异常: $e")
}

// 结果：
// I/System.out: 接收值 1
// I/System.out: 接收值 2
// I/System.out: 收到了异常: java.lang.IllegalStateException: 2 大于1

// 解释：
// 收到2的时候就抛出了异常，让后flow被取消，异常被捕获

通过catch函数

catch函数能够捕获之前产生的异常，之后的异常无法捕获。

flow {
    emit(1)
    emit(2)
    emit(3)
    emit(4)
    }.map {
        check(it <= 1) { "$it 大于1" }
        it
    }
    .catch { e -> println("Caught $e") }
    .collect()

// 结果：
// Caught java.lang.IllegalStateException: 2 大于1

Flow的取消

Flow的取消可以执行CoroutineScope.cancel

GlobalScope.launch {
    val flow1 = flow {
        emit(1)
        emit(2)
        emit(3)
        emit(4)
    }
    flow1.collect { value ->
        println("$value")
        if (value >= 3) {
            cancel()
        }
    }
}
        
 // 结果：1 2 3

StateFlow/MutableStateFlow

我们了解了Flow是冷信号，类似于RxJava中的observables。Kotlin还提供了一个类似于RxJava中的Subject的StateFlow。MutableStateFlow是相当于值可变的StateFlow

它有如下几个特点：

StateFlow主要用来管理和表示几种不同的状态

例如：网络请求中(Requesting)，网络请求完成(Request Complete)，网络请求失败(Request Fail)。

StateFlow只有值变化后才会释放新的值，和distinctUntilChanged类似
collect对于它不是必需的，StateFlow创建的时候就能开始释放值

class CounterModel {
    private val _counter = MutableStateFlow(0)  // 私有使用MutableStateFlow
    val counter = _counter.asStateFlow()   // 对外公开只读的StateFlow

    fun inc() {
        _counter.value++ //更改值
    }
}

val aModel = CounterModel()
val bModel = CounterModel()
val sumFlow: Flow<Int> = aModel.counter.combine(bModel.counter) { a, b -> a + b }

// 两个计时器的结果相加

本文主要讲解了Kotlin Flow的基础知识，下一篇我们将来介绍Kotlin Flow在UI界面编写，网络数据请求和数据库中是如何使用的。

最后

以上就是缥缈方盒为你收集整理的JetPack知识点实战系列十二：Kotlin Flow基础知识详解的全部内容，希望文章能够帮你解决JetPack知识点实战系列十二：Kotlin Flow基础知识详解所遇到的程序开发问题。

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

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。

本文分类：Kotlin
浏览次数：69 次浏览
发布日期：2023-08-17 05:10:04
本文链接：https://www.kaopuke.com/article/k-p-k_14_uzo_18_fx_13_j_26_3.html

JetPack知识点实战系列十二：Kotlin Flow基础知识详解

概述

Flow 引入的场景

Flow 的特性

Flow的构造函数

Flow中间运算函数

Delay相关的运算函数

debounce

Distinct相关的运算函数

distinctUntilChanged

Emitters相关的运算函数

transform

onStart

onCompletion

Limit相关的运算函数

drop

dropWhile

take

takeWhile

CoroutineContext相关的运算函数

flowOn

buffer

conflate

Flatten相关的运算函数

flatMapConcat

flattenConcat

flatMapMerge

flatMapMerge

merge

transformLatest

flatMapLatest

mapLatest

Transform相关的运算函数

filter

filterNot

filterIsInstance

filterNotNull

map

mapNotNull

withIndex

onEach

scan

runningReduce

Combine相关的运算函数

Flow结束函数

Collect相关的结束函数

collect

launchIn

collectIndexed

collectLatest

Collection相关的结束函数

toList

toSet

Count相关的结束函数

count

Reduce相关的结束函数

reduce

fold

single

singleOrNull

first/firstOrNull

Flow的错误异常处理

Flow的取消

StateFlow/MutableStateFlow

最后

相关文章

评论列表共有 0 条评论

发表评论 取消回复

发表评论取消回复