Flow 和 RxJava 差不多,不过 Flow 是和协程一起使用的 API。
简单的例子
flow {
emit(1)
emit(2)
}.collect {
println(it)
}
在 Flow 中可以使用 emit
发送数据,相当于 RxJava 中的 onNext()
。
创建 Flow
Flow 可以使用函数,集合,数组等来进行创建。
-
flowOf()
flowOf(1, 2, 3, 4, 5) .onEach { delay(it * 100L) } .collect { println(it) }
-
asFlow
listOf(1, 2, 3).asFlow() .collect { println(it) } arrayOf('a', 'b', 'c') .asFlow() .collect { println(it) }
asFlow
可以将任何使用Iterator
和数组转换为Flow
。{ 1 + 1 } .asFlow() .collect { println(it) } suspend { delay(1000) 1 + 1 } .asFlow() .collect { println(it) }
asFlow
还可以将函数(挂起函数也可以)的结果转换为flow
。 -
channelFlow
channelFlow { for (i in 1..5) { delay(i * 100L) send(i) } }.collect { println(it) }
channelFlow
是通过协程中的 channel 来实现的,所以它的发送数据和普通的flow
不太一样。channelFlow
是通过send
来发送的。flow
是冷的流。在没有切换协程的情况下,生产者和消费者是同步非阻塞的。channel
是热流。channelFlow
实现了生产者和消费者的异步非阻塞模型。
切换协程
Flow
是专用于 Kotlin 的 API,所以在会对协程进行切换,而不是协程。
flow {
for (i in 1..5) {
delay(100L)
emit(i)
}
}
.map {
it * it
}
.flowOn(Dispatchers.IO)
.collect {
println(it)
}
相比于 RxJava,flow 切换协程更加简单,只需要调用 flowOn
即可切换协程。
Flow 的取消
如果 Flow 是在一个挂起函数中被挂起了,那么 Flow 是可以取消的,否则不可以取消。
Terminal flow operator
Flow 的 API 和 Java 的相似,都有 Intermediate Operations、Terminal Operations。
Flow 的 Terminal 运算符可以是 suspend
函数,如 collect、single、reduce、toList 等,也可以是 launchIn 运算符,用户在指定的 CoroutineScope 内使用 flow。
整理一下 Flow 的 terminal operator,如下:
- collect
- single/first
- toList/toSet/toCollection
- count
- fold/reduce
- launchIn/produceIn/broadcastIn
collect
接受 flow 下数据流:
flow {
emit(1)
emit(2)
}.collect {
println(it)
}
single
如果 flow 中只发送一个数据可以通过 single 来返回,但是如果多于一个数据就会抛出 java.lang.IllegalArgumentException: Flow has more than one element
。
println("------flow terminal single")
val singleValue = flowOf(1)
.single()
println("single value = $singleValue")
// flowOf(1, 2)
// .single()
single 的源码如下:
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
}
first
获取 Flow 中的第一个元素。
val firstValue = flowOf(1, 2, 3)
.first()
println("first value = $firstValue")
first 的源码如下:
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
}
toList
toList 把 flow 中的数据添加到集合中。
val list = flowOf(1, 2, 3)
.toList()
println("list = $list")
val myList = mutableListOf(100, 200)
val myList2 = flowOf(1, 2, 3)
.toList(myList)
println("add myList as param res = $myList2")
toList 源码如下:
public suspend fun <T> Flow<T>.toList(destination: MutableList<T> = ArrayList()): List<T> = toCollection(destination)
public suspend fun <T, C : MutableCollection<in T>> Flow<T>.toCollection(destination: C): C {
collect { value ->
destination.add(value)
}
return destination
}
fold
用初始值作为初累计结果,然后再将累计的结果和每一个值做 operator 运算,然后将结果作为累计,用于下一次的计算。
Accumulate value starting with initial value and applying operation current accumulator value and each element.
val sum = flowOf(1, 2, 3)
.fold(0) { acc, value ->
acc + value
}
println("use flow fold to sum, result = $sum")
fold 源码如下:
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
}
reduce
reduce 和 fold 相似,不同的是 reduce 没有初始值。
val sum = flowOf(1, 2, 3)
.reduce { accumulator, value ->
accumulator + value
}
println("use flow reduce to sum, result = $sum")
reduce 源码如下:
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
}
launchIn
launchIn 会在执行的协程内调用 Flow 的 collect()
方法。
val simpleFlow: Flow<Int> = flow {
for (i in 1..10) {
emit(i)
println("launchIn ${currentCoroutineContext()[CoroutineName]?.name} emit $i")
delay(100)
}
}
coroutineScope {
launch(CoroutineName("SimpleCoroutine")) {
simpleFlow.launchIn(this)
}
}
launchIn 的源码如下:
public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = scope.launch {
collect() // tail-call
}