Kotlin中Flow操作符解析大全-五八三

在Kotlin中，Flow是一种异步数据流处理的声明式编程工具。它是一种冷流（cold stream）的概念，类似于 RxJava 中的 Observable，但有一些重要的区别。

本文基于Kotlin 1.8.20，协程 1.7.1

implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.1")

开头

简单的使用入门在前面的文章中我们有所介绍，具体请看：Flow流知识梳理

本次我们主要介绍Flow常用的操作符使用和原理，下面先看下操作符表格，了解下每个操作符的作用。

asFlow()	将Rang或者List等转换为Flow
map()	映射一次Flow发射的数据，只能变换发射的值
transform()	类似于Map，但是它可以控制Flow的发射
take()	控制Flow发射的数量，比如原始emit 3个，take(1)可以限制只发射1个
combine()	组合两个Flow，组合出来的Flow数据取决于最大的那个
zip()	组合两个Flow，组合出来的Flow数据取决去最小的那个
collectLatest()	接收最新的数据
buffer()	控制Flow缓冲区的大小，可平衡发送者和接受者之间的速度差异
conflate()	加入缓冲区，并且只接收最新的数据
flatMapConcat()	转换原始Flow发射出来的数据，并且转换出来的是另外一个Flow，它连接了原始和新的Flow
flatMapMerge()	它和flatMapConcat()作用是一致的，区别在于它并不能保证时序
flatMapLatest()	也是转换原始Flow，并且只接受原始Flow最新的数据
flowOn()	限定Flow发射的协程域

asFlow()

asFlow()可以将数组、列表和Lambda函数直接转换为一个Flow，比如我们我发射列表中的所有数据，不必在flowOf{}循环去emit每一个数据，直接使用asFlow()方法即可。

fun main() = runBlocking {







    createAsFlow().collect {
        println("collect asFlow $it")
    }


    createListAsFlow().collect {
        println("collect listAsFlow $it")
    }


}





// 将一个Rang对象转换成Flow对象
fun createAsFlow() = (0..2).asFlow()



// 将一个List对象转换成Flow对象
fun createListAsFlow() = listOf(1, 2, 3).asFlow()

# 
collect asFlow 0
collect asFlow 1
collect asFlow 2
collect listAsFlow 1
collect listAsFlow 2
collect listAsFlow 3

map()

map的意思是映射，它可以将Flow的原始数据映射成我们所需要的数据格式，它只可以改变发射出来的数据，比如我们将收集到的整数转换成字符串，就可以使用map()

fun main() = runBlocking {








    (0..2).asFlow().map { "Map一下 $it" }.collect {
        println("map collect $it")
    }
}





#

map collect Map一下 0
map collect Map一下 1
map collect Map一下 2

transform()

transform的意思是转换、变换，它可以将原始的Flow变换成另外一个Flow，它和map相同的是都可以改变Flow发射的数据，但是它的不同点在于它所变换的对象是Flow，而map映射的对象是流发射出来的数据，下面我们看下transform的具体用法和效果

fun main() = runBlocking {


    (0..1).asFlow().transform {
        emit(it + 1)
    }.collect {
        println("transform collect $it")
    }


}





#
transform collect 1
transform collect 2

示例代码中中标红的地方就可以看出，transform可以变换原始流，将原始流的数据转换成我们需要的数据再次发射emit出去

take()

take就是拿走、带走的意思，在Flow中它的作用是控制了接收的数量，比如说原始流要发射4个数据，但是我们只想要流中的前两个数据，那么可以直接使用take(2)来控制此数量

fun main() = runBlocking {







    (0..3).asFlow().take(2)
        .collect {
            println("collect take flow: $it")
        }
}





#

collect take flow: 0
collect take flow: 1

combine()

combine是结合的意思，在Flow中它可以将两个流发射出来的数据结合在一起

fun main() = runBlocking {







    val stringFlow = listOf("a", "b").asFlow()
    (0..2).asFlow().combine(stringFlow) { origin1, origin2 ->
        "$origin1 - $origin2"
    }.collect {
        println("collect combine flow: $it")
    }


}





# 
collect combine flow: 0 - a
collect combine flow: 1 - b
collect combine flow: 2 - b

注意看上面的代码，stringFlow只会发射两个数据，(0..2)会发射三个数据，通过combine()组合只会一共会发射三个数据，它是以最多数据的那个流为主，超出的数据都是和另一个流的最后一个数据来组合

zip()

zip也是组合的意思，它和combine的作用是一致的，都是将两个流数据组合起来，但是最终效果两者有所区别，zip最终的数据量是根据组合前两个流数量最少的那个为主

fun main() = runBlocking {







    createZip1Flow().zip(createZip2Flow()) { i, j ->
        "$i - $j"
    }.collect {
        println("collect zip $it")
    }


}





fun createZip1Flow() = (0..2).asFlow()

fun createZip2Flow() = listOf("a", "b").asFlow()



#
collect zip 0 - a
collect zip 1 - b

zip1Flow本身会发射三个数据，而zip2Flow只会发射两个数据，使用zip之后，我们最终得到的只有两个数据，zip1Flow中第三个数据会被舍弃。

collectLatest()

latest字面意思是最新的、最近的意思，collectLatest联想一下是不是收集最新最近的数据呢？我们通过代码来验证下此想法

fun main() = runBlocking {







    createLatestFlow().collectLatest {
        println("start collectLatest: $it")
      	// 模拟耗时操作500毫米
        delay(500L)
        println("end collectLatest: $it")
    }


}





fun createLatestFlow() = flow {
    var i = 0
    while (i < 10) {
        emit(i++)
      	// 发射完数据之后延时100毫秒
        delay(100)
    }

}

先看下上面代码，createLatestFlow每次发射完数据之后都会延时100毫秒，并且只发射10次，10次之后停止发射；collectLatest有两处日志打印，start处是延时之前，end是延时之后。其实一开始接触collectLatest的时候我的预期是每过500毫秒接收一次最新的数据，然而事实却并非如此，collectLatest的实际情况是接收到一个数据的时候，延时之前（start处）会得到正常的打印，而延时之后（end处）的打印却不会执行，因为在延时500毫秒之后，它已经接收到了另外一个或多个新的数据，所以它会抛弃旧数据，也就是end打印不会再执行了，等到流发射完最后一个数据的时候，没有新数据发射了，end会执行最后一个接收到的数据。

也就是说start处每次接收到数据的时候都会得到输出，而end只会输出最后一个数据，下面执行看看输出是不是如此

start collectLatest: 0
start collectLatest: 1
start collectLatest: 2
start collectLatest: 3
start collectLatest: 4
start collectLatest: 5
start collectLatest: 6
start collectLatest: 7
start collectLatest: 8
start collectLatest: 9
end collectLatest: 9

通过输出就可以论证我们上面的分析确实没错。如果讲collectLatest模拟延时时间改为50毫秒，小于发射延时呢？效果会是咋样，接收处理的延时小于发射的延时，在接收的函数中start和end应该是都可以得到正常的输出。

fun main() = runBlocking {
    createLatestFlow().collectLatest {
        println("start collectLatest: $it")
        delay(50L)
        println("end collectLatest: $it")
    }


}





fun createLatestFlow() = flow {
    var i = 0
    while (i < 10) {
        emit(i++)
        delay(100)
    }
}

#
start collectLatest: 0
end collectLatest: 0
start collectLatest: 1
end collectLatest: 1
start collectLatest: 2
end collectLatest: 2
start collectLatest: 3
end collectLatest: 3
start collectLatest: 4
end collectLatest: 4
start collectLatest: 5
end collectLatest: 5
start collectLatest: 6
end collectLatest: 6
start collectLatest: 7
end collectLatest: 7
start collectLatest: 8
end collectLatest: 8
start collectLatest: 9
end collectLatest: 9

buffer()

一看到buffer函数，基本就可以联想到缓冲了，也就是经典的背压问题，当发射处流速大于接收处速度时就会产生背压问题，我们可以通过上面collectLatest()函数来只接收最新数据，当然如果我们想接收每一个数据时就可以通过buffer来处理了。

下面我们模拟下接收处比发射处要慢的场景，看看流的运行是什么样的

fun main() = runBlocking {


    val time = System.currentTimeMillis()

    flow {

        for (i in 1..10) {

            // 模拟每次延时 1 秒发射数据

            delay(1000L)

            emit(i)

        }


    }.onEach {

        println("emit $it, time: ${System.currentTimeMillis() - time}")

    }.collect {
        // 模拟每次接收的时候都延时 2 秒

        delay(2000L)

        println("buffer flow collect $it, time: ${System.currentTimeMillis() - time}")

    }

}

在上面示例中，我们通过 delay将接收的延时设置大于发射延时，看看输出的情况

emit 1, time: 1028
buffer flow collect 1, time: 3033
emit 2, time: 4035
buffer flow collect 2, time: 6040
emit 3, time: 7045
buffer flow collect 3, time: 9045
emit 4, time: 10050
buffer flow collect 4, time: 12055
emit 5, time: 13056
buffer flow collect 5, time: 15057
emit 6, time: 16059
buffer flow collect 6, time: 18061
emit 7, time: 19062
buffer flow collect 7, time: 21066
emit 8, time: 22072
buffer flow collect 8, time: 24074
emit 9, time: 25077
buffer flow collect 9, time: 27079
emit 10, time: 28080
buffer flow collect 10, time: 30081

从输出日志中可以看出，发射+接收此 10 个数据一共花费了 30s，流的每次发射需要等待接收处理完成才会发射下一个数据，这就导致了总共花费的时间额外的长，试想一下如果在业务处理中那么会白白的浪费我们等待的时间，此时我们就可以通过 buffer()来解决这种问题，接下来我们再看看加上 buffer()的效果是咋样的

fun main() = runBlocking {


    val time = System.currentTimeMillis()

    flow {

        for (i in 1..10) {

            // 模拟每次延时 1 秒发射数据

            delay(1000L)

            emit(i)

        }


    }.onEach {

        println("emit $it, time: ${System.currentTimeMillis() - time}")

    }.buffer().collect {
        // 模拟每次接收的时候都延时 2 秒

        delay(2000L)

        println("buffer flow collect $it, time: ${System.currentTimeMillis() - time}")

    }

}

在原有的代码上仅仅加上了 buffer()函数，未改变其它地方，再看看日志输出

emit 1, time: 1051
emit 2, time: 2057
buffer flow collect 1, time: 3057
emit 3, time: 3058
emit 4, time: 4059
buffer flow collect 2, time: 5059
emit 5, time: 5060
emit 6, time: 6065
buffer flow collect 3, time: 7059
emit 7, time: 7066
emit 8, time: 8071
buffer flow collect 4, time: 9060
emit 9, time: 9073
emit 10, time: 10074
buffer flow collect 5, time: 11061
buffer flow collect 6, time: 13065
buffer flow collect 7, time: 15070
buffer flow collect 8, time: 17071
buffer flow collect 9, time: 19072
buffer flow collect 10, time: 21077

从此输出日志中可以看出，发射+接收此 10 个数据一共花费了 21s，并且可以看出此时发射数据不再受制于接收处理完了再发射下一个数据，buffer()为 Flow提供了一个缓冲区，它将接收到的数据缓冲起来，这样发射处就不用等待接收处处理完数据再进行数据的发射，也就可以减少 Flow执行的总体时间。

buffer()有一个可选的参数capacity，它表示了缓冲区的大小，默认为Channel.BUFFERED也就是 64

UNLIMITED表示不限制缓冲区的大小，可为无限大，但是此模式占用的内存大，慎用；
RENDEZVOUS表示缓冲区大小为 0，发射处在遇到接收处处理数据的时候，必须等待接收处处理完了才会发射新的数据；
BUFFERD是默认的模式，表示缓冲区大小为固定的 64；
CONFLATED这个模式表示每次接收的时候只会取最新的那个数据，之前未来得及处理的数据都直接抛弃不再处理。

conflate()

conflate字面意思是合并，但是看源码其实就是调用buffer(CONFLATED)函数，用于添加缓冲区并且只处理最新的数据，此处就不再过多解析。

flatMapConcat()

flatMapConcat()包括下面介绍的 flatMapMerge()和flatMapLatest()函数都是转换原始流数据的作用，并且在转换之后做了额外的操作，

我们先看下flatMapConcat()接收的参数：transform: suspend (value: T) -> Flow<R>，参数是一个 Lambda，并且返回的是一个 Flow对象，也就是说它将原始流的数据接收到之后再转换成了一个新的流，最终将新的流数据发射出去，下面我们来实践看看效果

@OptIn(ExperimentalCoroutinesApi::class)

fun main() = runBlocking {

    createFlatMapFlow().flatMapConcat {
      	// 这里需要返回一个 Flow 对象，我们创建了一个新的 Flow，并且原封不动的将原始流数据再发射出去
        flow {
            emit(it)
        }
    }.collect {
        println("collect flatMapConcat $it")
    }
}



fun createFlatMapFlow() = flow {
    (3 downTo 1).forEach {
        emit(it)
    }

}


#
collect flatMapConcat 3
collect flatMapConcat 2
collect flatMapConcat 1

看到上面代码是不是有种疑问，这函数有啥用啊，原始的流不要，新建一个流替换它发射数据？这里大家仔细的看一下，flatMapConcat是对原始流的每一个数据都回新建一个流，然后发射，试想一种场景，先查询一个年级的所有班级信息，然后再查询每个班级的学生信息，这种场景是不是特别适合flatMapConcat操作符，它先接收每个班级的信息，然后对班级中每个学生信息再进行查询

@OptIn(ExperimentalCoroutinesApi::class)

fun main() = runBlocking {

    fetchClassInfo().flatMapConcat {
        fetchStudent(it)
    }.collect {}
}





fun fetchClassInfo() = flow {
    emit(""/*班级信息*/)
}

fun fetchStudent(classInfo: String) = flow {
    emit(""/*学生信息*/)
}

flatMapMerge()

如果把上面flatMapConcat()换成flatMapMerge()，我们会发现输出的日志竟然一模一样，确实如此两者在这方面的作用是一致的，文章开头的表格中我们也提到了，flatMapMerge()的区别在于时序上面不能保证原始流的发射顺序，下面我们通过模拟耗时操作来看下两者在时序上面的区别

@OptIn(ExperimentalCoroutinesApi::class)

fun main() = runBlocking {

    createFlatMergeFlow().flatMapConcat {
        flow {

            // 模拟耗时操作，根据原始流数据*100 毫秒，原始流发射数据为3，2，1
            kotlinx.coroutines.delay(it * 100L)
            emit(it)
        }


    }.collect {
        println("collect flatMapConcat $it")
    }


    createFlatMergeFlow().flatMapMerge {
        flow {
            // 模拟耗时操作，根据原始流数据*100 毫秒，原始流发射数据为3，2，1
            kotlinx.coroutines.delay(it * 100L)
            emit(it)
        }
    }.collect {
        println("collect createFlatMergeFlow $it")
    }
}

fun createFlatMergeFlow() = flow {
    (3 downTo 1).forEach {
        emit(it)
    }
}

上述代码我们原始流的发射数据应该是 3，2，1，然后通过flatMapConcat和flatMapMerge内部同样的逻辑来转换流的发射，以此延时 300ms、200ms、100ms，如果两者都能保证原始流的时序，那么是不是转换之后还是应该都输出3，2，1呢？运行看下具体的输出

collect flatMapConcat 3
collect flatMapConcat 2
collect flatMapConcat 1
collect createFlatMergeFlow 1
collect createFlatMergeFlow 2
collect createFlatMergeFlow 3

通过日志就可以清楚的知道，在延时的作用下flatMapConcat依旧可以保证原始流的发射顺序，然而flatMapMerge却改变了原始流的发射顺序，延时少的先得到输出，延时越长发射越慢，这就是两者在时序上的区别。

flatMapLatest()

flatMapLatest其实就是在转换原始流的操作了结合了collectLatest的特征，它只接收原始流最新的数据，并且将最新数据转换成另外一个流。

这里就简单看个示例

@OptIn(ExperimentalCoroutinesApi::class)

fun main() = runBlocking {

    createFlatMapLatestFlow().flatMapLatest {
        flow {

            println("start flatMapLatest: $it")
            delay(300L)
            println("end flatMapLatest: $it")
            emit(it)
        }
    }.collect {
        println("collect flatMapLatest: $it")
    }
}

fun createFlatMapLatestFlow(): Flow<Int> {
    return flow {
        (0..10).forEach {
            delay(100L)
            emit(it)
        }
    }
}

#
start flatMapLatest: 0
start flatMapLatest: 1
start flatMapLatest: 2
start flatMapLatest: 3
start flatMapLatest: 4
start flatMapLatest: 5
start flatMapLatest: 6
start flatMapLatest: 7
start flatMapLatest: 8
start flatMapLatest: 9
start flatMapLatest: 10
end flatMapLatest: 10
collect flatMapLatest: 10

原始流每次延时 100ms 发射数据，然后在flatMapLatest()中模拟了 300ms 的延时操作，300ms 之前的日志每次都是可以得到输出，300ms 之后只有最新的数据 10 才可以得到输出，而且在最新的收集中也只有最新的数据 10，看到这里是不是就是类似flatMapConcat结合了collectLatest两者的特性，还是比较容易理解的。

flowOn()

flowOn()函数主要就是用来指定流发射时的协程环境，如果我们想指定发射的操作在子线程中，无需通过thread或者 CoroutinesScope来切换线程，只需要调用 flowOn(Dispatchers.Default/IO)来切换即可，下面通过示例来看看实际效果

fun main() = runBlocking {







    createFlowOn().collect {

        println("flowOn collect $it, thread ${Thread.currentThread().name}")

    }


}



fun createFlowOn() = flow {

    for (i in 0..2) {

        emit(i)

        println("emit thread ${Thread.currentThread().name}")

    }


}


#

flowOn collect 0, thread main
emit thread main
flowOn collect 1, thread main
emit thread main
flowOn collect 2, thread main
emit thread main

此时我们未调用 flowOn()函数来切换线程，通过日志可以直观的看出，Flow的发射和接收是在同一个线程中，也就是main线程中，然后我们再给它指定个 IO协程环境看看

fun main() = runBlocking {







    createFlowOn().collect {

        println("flowOn collect $it, thread ${Thread.currentThread().name}")

    }


}



fun createFlowOn() = flow {

    for (i in 0..2) {

        emit(i)

        println("emit thread ${Thread.currentThread().name}")

    }


}.flowOn(Dispatchers.IO)


#

emit thread DefaultDispatcher-worker-1
emit thread DefaultDispatcher-worker-1
emit thread DefaultDispatcher-worker-1
flowOn collect 0, thread main
flowOn collect 1, thread main
flowOn collect 2, thread main

加上 flowOn(Dispatcher.IO)之后，Flow的发射处就变成了子线程，接收依旧在主线程中，这样它的效果就很明显了。

到此为止， Flow的大部分操作符就介绍结束了，如果文章中有什么你觉得不对的地方帮忙指出，谢谢！

文章版权归作者所有，未经允许请勿转载，侵权请联系 admin@trc20.tw 删除。

THE END