前语
Flow
是kotlin协程中的流。RxJava便是流式编程的库。Flow归于冷流对应RxJava中的Observable Flowable Single MayBe和Completable等。Kotlin协程中的暖流完成MutableSharedFlow和MutableStateFlow等,对应RxJava中暖流PublisherSubject和BehaviorSubject。
- 冷流:较少的拜访和修正
- 暖流:频繁地读取和更新
Flow运用
fun main() {
runBlocking (Dispatchers.Default){
// 发送10个元素,从0到9
val myFlow = flow {
repeat(10){
emit(it)
}
}
launch {
myFlow.collect{
println("Coroutine1:$it")
}
}
launch {
myFlow.collect{
println("Coroutine2:$it")
}
}
}
}
协程1和2经过Flow.collect
订阅Flow。
fun main() {
runBlocking (Dispatchers.Default){
// 发送10个元素,从0到9
val myFlow = flow {
repeat(10){
// 修正本来的CoroutineContext,会反常
withContext(Dispatchers.IO){
emit(it)
}
}
}
launch {
myFlow.collect{
println("Coroutine1:$it")
}
}
}
}
Flow约束,不能修正本来的CoroutineContext。能够运用ChannelFlow
就能正常运用。
fun main() {
runBlocking (Dispatchers.Default){
// 发送10个元素,从0到9
val myFlow = channelFlow {
repeat(10){
// 能够修正本来的CoroutineContext
withContext(Dispatchers.IO){
channel.send(it)
}
}
}
launch {
myFlow.collect{
println("Coroutine1:$it")
}
}
}
}
fun main() {
runBlocking(Dispatchers.Default) {
// 发送10个元素,从0到9
val myFlow = flow {
repeat(10) {
try {
emit(it)
} catch (e: Throwable) {
emit(22)
}
}
}
launch {
myFlow.collect {
if (it == 2) {
// 这儿出现反常后,collect订阅就完毕了(只打印2次,第三次就反常了)
error("Error")
}
println("Coroutine1:$it")
}
}
}
}
Flow中collect反常,那么订阅就完毕了。
Flow作业原理
public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)
private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
override suspend fun collectSafely(collector: FlowCollector<T>) {
collector.block()
}
}
public final override suspend fun collect(collector: FlowCollector<T>) {
val safeCollector = SafeCollector(collector, coroutineContext)
try {
collectSafely(safeCollector)
} finally {
safeCollector.releaseIntercepted()
}
}
创立SafeFlow,承继于AbstractFlow,订阅调用的是collect。查看当前CoroutineContext和调用的collect方法传入的是否一致,不一致就抛出反常。
Flow被SafeCollector代理去查看反常。
转换前的流称上游Upstream
,处理后再发送到下流Downstream
flatMap操作符
类似于RxJava中的concatMap操作符
fun main() {
runBlocking(Dispatchers.Default) {
val myFlow = flow {
repeat(3) {
emit(it)
}
}
launch {
// 将本来的流元素构建成一个新的流(按照本来的流元素输出)
myFlow.flatMapConcat { upstreamValue ->
flow {
delay(1000L - upstreamValue * 100)
repeat(2) {
emit(upstreamValue * 10 + it)
}
}
}.collect {
println("collect $it")
}
}
}
}
输出:
collect 0
collect 1
collect 10
collect 11
collect 20
collect 21
将本来发送3个元素,经过flatMapConcat()
发送两个元素。越是先发送的元素延迟时间越长,然后按次序输出6个元素。
flatMapMerge
类似于RxJava中的flatMap
fun main() {
runBlocking(Dispatchers.Default) {
val myFlow = flow {
repeat(3) {
emit(it)
}
}
launch {
// 将本来的流元素构建成一个新的流(默认并发16个)谁先履行完就发送谁
myFlow.flatMapMerge { upstreamValue ->
flow {
delay(1000L - upstreamValue * 100)
repeat(2) {
emit(upstreamValue * 10 + it)
}
}
}.collect {
println("collect $it")
}
}
}
}
输出:
collect 20
collect 21
collect 10
collect 11
collect 0
collect 1
不会保证本来的次序,哪个流先处理完就先发送数据。concurrency
默认值16,并行履行的数量。当concurrency为1时和flatMapConcat一样。
fun main() {
runBlocking(Dispatchers.Default) {
val myFlow = flow {
repeat(3) {
emit(it)
}
}
launch {
// 将本来的流元素构建成一个新的流(并发数是2,到达2个的时候等候然后再履行下一个)
myFlow.flatMapMerge(2) { upstreamValue ->
flow {
delay(1000L - upstreamValue * 100)
repeat(2) {
emit(upstreamValue * 10 + it)
}
}
}.collect {
println("collect $it")
}
}
}
}
输出:
collect 10
collect 11
collect 0
collect 1
collect 20
collect 21
flatMapLatest
类似于RxJava中的switchMap
fun main() {
runBlocking(Dispatchers.Default) {
val myFlow = flow {
repeat(3) {
emit(it)
}
}
launch {
// 前面没履行完的Flow会被撤销,然后被后续的Flow替换
myFlow.flatMapLatest { upstreamValue ->
flow {
delay(1000L - upstreamValue * 100)
repeat(2) {
emit(upstreamValue * 10 + it)
}
}
}.collect {
println("collect $it")
}
}
}
}
输出:
collect 20
collect 21
总结
介绍了Flow常用的操作符map flatMap(串式) flatMapMerge(并发) flatmapLatest(替代旧的)等简略运用。