flat登录html模板酷站登陆(h5酷站)

本文作者

作者：搬砖小子出现了

链接：

Kotlin Flow 基本上可以替代RxJava，其提供了诸多操作符来处理数据。本篇文章为其分一下类，演示下基本用法。

恩...建议大家看一眼右侧的滚动条，然后点击下收藏。

弃用的方法就不罗列了。Ctrl + F 快速查找

前置资料

冷流 ?

冷指的是无消费者时则不会生产数据。

热流 ?

热指的是无观察者时也会生产数据。

Flow分类

一般 Flow

一般的Fow , 仅有一个观察者。冷流。

//构建

val testFlow = flow<String>{

emit( "hello")

emit( "flow")

}

//接收

coroutineScope.launch{

testFlow.collect{ value->

println( value)

}

//打印

hello

flow

StateFlow

有状态的Flow ，可以有多个观察者，热流

构造时需要传入初始值 : initialState

常用作与UI相关的数据观察，类比LiveData

//创建

val uiState=MutableStateFlow(Result.Loading)

//监听

coroutineScope.launch{

uiState.collect{ value->

println( value)

}

//赋值

uiState. value=Result.Sucess

//打印结果

Result.Loading

Result.Sucess

SharedFlow

可定制化的StateFlow，可以有多个观察者，热流. 无需初始值，有三个可选参数：

extraBufferCapacity- 除了replay之外缓冲的值的数量。当有剩余缓冲区空间时， emit不会挂起（可选，不能为负，默认为零）。

onBufferOverflow- 配置缓冲区溢出的操作（可选，默认为暂停尝试发出值）

使用SharedFlow 你可以写个 FlowEventBus。

//创建

val signEvent=MutableSharedFlow <String>

//监听

coroutineScope.launch{

signEvent.collect{ value->

println( value)

}

//赋值

signEvent.tryEmit( "hello")

signEvent.tryEmit( "shared flow")

//打印结果

hello

shared flow

操作符

• 中间操作符

一般来说是用来执行一些操作，不会立即执行，返回值还是个Flow。

• 末端操作符

会触发流的执行，返回值不是Flow。

创建Flow

flow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow.html

创建Flow的基本方法.

使用 emit 发射单个值

使用 emitAll 发射一个流，类似 list.addAll(anotherList)

flow < Int> {

emit(1)

emit(2)

emit(flowOf(1,2,3))

}

flowOf

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow-of.html

快速创建 flow ,类比listOf。

val testFLow = flowOf( 1, 2, 3)

launch{

testFLow.collect{ value->

print( value)

}

//打印结果

asFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/as-flow.html

将其他数据转换成普通的flow ，一般是集合向Flow的转换。

listOf(1,2,3) .asFlow

callbackFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/callback-flow.html

将回调方法改造成flow ,类似suspendCoroutine。

funflowFrom(api: CallbackBasedApi) : Flow<T> = callbackFlow {

valcallback = object: Callback {

overridefunonNextValue(value: T) {

send(value)

.onFailure { throwable ->

}

overridefunonApiError(cause: Throwable) {

cancel(CancellationException( "API Error", cause))

}

overridefunonCompleted= channel.close

}

api.register(callback)

awaitClose { api.unregister(callback) }

}

emptyFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/empty-flow.html

返回一个空流。

emptyFlow < Int>

channelFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/channel-flow.html

在一般的flow在构造代码块中不允许切换线程，ChannelFlow则允许内部切换线程。

//构建

val channelFlow = channelFlow<String> {

send( "hello")

withContext(Dispatchers.IO) {

send( "channel flow")

}

//监听

coroutineScope.launch{

signEvent.collect{ value->

println( value)

}

末端操作符

collect

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/collect.html

触发flow的运行。通常的监听方式。

launch{

flowOf( 1, 2, 3).collect{ value->

print( value)

}

// 1 2 3

复制代码

collectIndexed

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/collect-indexed.html

带下标的收集操作

launch{

flowOf( 1, 2, 3).collectIndexed{ value->

print( value)

}

// 1 2 3

collectLatest

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/collect-latest.html

与 collect的区别是，有新值发出时，如果此时上个收集尚未完成，则会取消掉上个值的收集操作。

flow {

emit( 1)

delay( 50)

emit( 2)

} .collectLatest { value ->

println( "Collecting $value")

delay( 100) // Emulate work

println( "$value collected")

}

//输出

Collecting 1

Collecting 2

2collected

只想要最新的数据，中间值可以丢弃时可以使用此方式。

toCollection

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/to-collection.html

将结果添加到集合。

val array= arrayListOf( 0)

launch {

flow {

emit( 1)

emit( 2)

} .toCollection( array)

}

array. forEach{ value->

print(value)

}

//打印结果

012

toList

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/to-list.html

将结果转换为List。

flow {

emit( 1)

emit( 2)

} .toList.forEach{ value->

print( value)

}

// 1 2

toSet

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/to-set.html

将结果转换为Set。

flow {

emit( 1)

} .toSet.forEach{ value->

print( value)

}

// 1

launchIn

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/launch-in.html

直接触发流的执行，不设置action,入参为coroutineScope 一般不会直接调用，会搭配别的操作符一起使用，如onEach,onCompletion。返回值是Job。

flow {

emit(1)

emit(2)

}.launchIn ( lifecycleScope )

last

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/last.html

返回流发出的最后一个值 ,如果为空会抛异常。

val myFlow= flow {

emit( 1)

emit( 2)

}

launch{

print(myFlow.last)

}

// 2

lastOrNull

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/last-or-nul

返回流发出的最后一个值 ,可以为空。

valmyFlow= emptyFlow< Int>

launch{

print(myFlow.lastOrNull)

}

// null

first

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/first.html

返回流发出的第一个值 ,如果为空会抛异常。

val myFlow= flow {

emit( 1)

emit( 2)

}

launch{

print(myFlow.first)

}

// 1

firstOrNull

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/first-or-null.html

返回流发出的第一个值 ,可以为空。

valmyFlow= emptyFlow< Int>

launch{

print(myFlow.firstOrNull)

}

// null

single

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/single.html

接收流发送的第一个值，区别于first,如果为空或者发了不止一个值，则都会报错。

val myFlow= flow {

emit( 1)

}

launch {

print(myFlow.single) // 1

}

val myFlow1= flow {

emit( 1)

emit( 2)

}

launch {

print(myFlow 1. single ) // error

}

singleOrNull

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/single-or-null.html

接收流发送的第一个值，可以为空 ,发出多值的话除第一个，后面均被置为null。

val myFlow= flow {

emit( 1)

}

launch {

print(myFlow. singleOrNull ) // 1

}

count

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/count.html

返回流发送值的个数。类似list.size，注：sharedFlow无效(无意义）

val myFlow= flow {

emit( 1)

emit( 2)

}

launch{

print(myFlow.count)

}

//2

fold

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/fold.html

从初始值开始执行遍历,并将结果作为下个执行的参数。

val sum= flowOf( 2, 3, 4)

.fold( 1, { result, value->

result + value

})

// sum = 10，相当于 1 + 2 + 3 + 4

reduce

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/reduce.html

和fold 差不多，无初始值。

val result= flowOf( 1, 2, 3)

.reduce { acc, value->

acc + value

}

//result = 6 1 + 2 +3

回调操作符

onStart

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-start.html

在上游流开始之前被调用。可以发出额外元素,也可以处理其他事情，比如发埋点。

flow < Result> {

emit(Result.Success)

}.onStart{

emit(Result.Loading)

}

onCompletion

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-completion.html

在流取消或者结束时调用。可以执行发送元素，发埋点等操作。

flow < Result> {

emit(Result.Success)

}.onCompletion{

emit(Result.End)

}

onEach

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-each.html

在上游向下游发出元素之前调用。

flow<Int>{

emit( 1)

emit( 2)

emit( 3)

}.onEach{ value->

println( value)

}.launchIn(lifecycleScope)

// 打印结果

flat登录html模板酷站登陆(h5酷站),flat登录html模板酷站登陆(h5酷站),flat登录html模板酷站登陆,文章,模板,html,第1张

onEmpty

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-empty.html

当流完成却没有发出任何元素时回调。可以用来兜底.。

emptyFlow< String>.onEmpty {

emit( "兜底数据")

} .launchIn(lifecycleScope)

onSubion

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-subion.html

SharedFlow 专属操作符（StateFlow是SharedFlow 的一种特殊实现）

val state = MutableSharedFlow<String>.onSubion {

emit( "onSubion")

}

launch{

state.collect { value->

println( value)

}

//打印结果

onSubion

变换操作符

map

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map.html

将发出的值进行变换，lambda的返回值为最终发送的值。

flow {

emit( 1)

emit( 2)

} .map { value->

value* 2

} .collect {

println( value)

}

//打印结果

mapLatest

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map-latest.html

类比collectLatest ,当有新值发送时如果上个变换还没结束，会先取消掉。

flow {

emit( "a")

delay( 100)

emit( "b")

}.mapLatest { value->

println( "Started computing $value")

delay( 200)

"Computed $value"

}.collect { value->

print( value)

}

// 打印结果

Started computing a

Started computing b

Computed b

mapNotNull

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map-not-null.html

仅发送 map后不为空的值。

flow {

emit( "a")

emit( "b")

} .mapNotNull { value->

if( value!= "a") {

value

} else{

null

}

}.collect { value->

print( value)

}

// 结果

transform

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform.html

对发出的值进行变换。区别于map， transform的接收者是FlowCollector，因此它非常灵活，可以变换、跳过它或多次发送。

flow {

emit( 1)

emit( 2)

} .transform { value->

if( value== 1) {

emit( "value :$value*2")

}

emit( "transform :$value")

}.collect { value->

println( value)

}

// 打印结果

value: 1* 2

transform : 1

transform : 2

transformLatest

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform-latest.html

类比 mapLatest,当有新值发送时如果上个变换还没结束，会先取消掉。

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map-latest.html

flow {

emit( "a")

delay( 100)

emit( "b")

}.transformLatest { value->

emit( value)

delay( 200)

emit( value+ "_last")

}.collect { value->

println( value)

}

// 打印结果

b_last

transformWhile

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform-while.html

这个变化的lambda 的返回值是 Boolean ,如果为 False则不再进行后续变换, 为 True则继续执行。

flow {

emit( "a")

emit( "b")

} .transformWhile { value->

emit( value)

true

} .collect { value->

println( value)

}

//结果

--------------------

flow {

emit( "a")

emit( "b")

}.transformWhile { value->

emit( value)

false

}.collect { value->

println( value)

}

//结果

asStateFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/as-state-flow.html

将 MutableStateFlow转换为 StateFlow ，就是变成不可变的。常用在对外暴露属性时使用。

privateval_uiState = MutableStateFlow<UIState>(Loading)

valuiState = _uiState.asStateFlow

asSharedFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/as-shared-flow.html

将 MutableSharedFlow转换为 SharedFlow ，就是变成不可变的。常用在对外暴露属性时使用。

privateval_uiState = MutableStateFlow<UIState>(Loading)

valuiState = _uiState.asStateFlow

receiveAsFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/receive-as-flow.html

将Channel 转换为Flow ,可以有多个观察者，但不是多播，可能会轮流收到值。

privateval _event = Channel< Event>

val event= _event.receiveAsFlow

consumeAsFlow

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/consume-as-flow.html

将Channel 转换为Flow ,但不能多个观察者（会crash）!

privateval _event = Channel< Event>

val event= _event.consumeAsFlow

withIndex

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/with-index.html

将结果包装成IndexedValue类型。

flow {

emit( "a")

emit( "b")

} .withIndex.collect {

print(it.index + ": "+ it. value)

}

//结果

0: a

1: b

scan

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/scan.html

和 fold 相似，区别是fold 返回的是最终结果，scan返回的是个flow ，会把初始值和每一步的操作结果发送出去。

flowOf( 1, 2, 3).scan( 0) { acc, value->

acc + value

}.collect {

print(it)

}

// 0 1 3 6

acc 是上一步操作的结果， value是发射的值

0是初始值

1是 0+ 1= 1

3是 1+ 2= 3

6是 3+ 3= 6

produceIn

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/produce-in.html

转换为ReceiveChannel , 不常用。

注：Channel 内部有 ReceiveChannel 和SendChannel之分,看名字就是一个发送，一个接收。

flowOf( 1, 2, 3).produceIn( this)

.consumeEach { value->

print( value)

}

//1 2 3

runningFold

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/running-fold.html

区别于 fold ，就是返回一个新流，将每步的结果发射出去。

flowOf( 1, 2, 3).runningFold( 1){ acc, value->

acc + value

} .collect { value->

print( value)

}

// 1 2 4 7

runningReduce

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/running-reduce.html

区别于 reduce ，就是返回一个新流，将每步的结果发射出去。

flowOf( 1, 2, 3).runningReduce { acc, value->

acc + value

} .collect { value->

print( value)

}

// 1 3 6

shareIn

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/share-in.html

将普通flow 转化为 SharedFlow , 其有三个参数：

scope: CoroutineScope开始共享的协程范围。

started: SharingStarted 控制何时开始和停止共享的策略。

其中 started 有一些可选项:

Eagerly: 共享立即开始，永不停止。

WhileSubscribed 具有以下可选参数：

replayExpirationMillis- 共享的协程从停止到重新激活，这期间缓存的时效。

val share = flowOf( 1, 2, 3).shareIn( this,SharingStarted.Eagerly)

//可以有多个观察者

state.collect{ value->

print( value)

}

stateIn

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/state-in.html

将普通flow 转化为 StateFlow 。其有三个参数：

scope- 开始共享的协程范围。

started- 控制何时开始和停止共享的策略。

initialValue- 状态流的初始值。

val state = flowOf(Success).stateIn(lifecycleScope,SharingStarted.Eagerly,Loading)

state.collect{ value->

print( value)

}

// Loading Success

stateIn和sharedIn通常用在其他来源的flow的改造监听，不会像上面那样使用。

过滤操作符

filter

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/filter.html

筛选出符合条件的值。

flow {

emit( "a")

emit( "b")

}.filter { value->

value== "a"

}.collect { value->

print( value)

}

//结果

filterInstance

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/filter-is-instance.html

筛选对应类型的值。

flow {

emit( "a")

emit( "b")

emit( 1)

}.filterIsInstance<String>.collect { value->

print( value)

}

//结果

filterNot

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/filter-not.html

筛选不符合条件相反的值,相当于filter取反。

flow {

emit( "a")

emit( "b")

}.filterNot { it == "a"} .collect { value->

print( value)

}

//结果

filterNotNull

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/filter-not-null.html

筛选不为空的值。

flow {

emit( "a")

emit( null)

emit( "b")

}.filterNotNull.collect { value->

print( value)

}

//结果

drop

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/drop.html

入参count为int类型 ,作用是丢弃掉前 n 个的值。

flow {

emit( 1)

emit( 2)

emit( 3)

}.drop( 2).collect { value->

print( value)

}

//结果

dropWhile

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/drop-while.html

这个操作符有点特别，和filter 不同！它是找到第一个不满足条件的，返回其和其之后的值。

如果首项就不满足条件，则是全部返回。

flow {

emit( 3)

emit( 1) //从此项开始不满足条件

emit( 2)

emit( 4)

}. dropWhile { it == 3} .collect { value->

print( value)

}

//结果

124

flow {

emit( 1) //从首项开始就不满足条件

emit( 2)

emit( 3)

emit( 4)

}. dropWhile { it == 3} .collect { value->

print( value)

}

//结果

1234

take

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/take.html

返回前 n个元素。

flow {

emit( 1)

emit( 2)

emit( 3)

} .take( 2) .collect { value->

print( value)

}

//结果

takeWhile

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/take-while.html

也是找第一个不满足条件的项，但是取其之前的值，和dropWhile 相反。

如果第一项就不满足，则为空流。

flow {

emit( 1)

emit( 2)

emit( 3) //从此项开始不满足条件

emit( 4)

} .takeWhile { it < 3} .collect { value->

print( value)

}

//结果

flow {

emit( 3) //从此项开始不满足条件

emit( 1)

emit( 2)

emit( 4)

} .takeWhile { it < 3} .onEmpty {

print( "empty")

}.collect { value->

print( value)

}

//结果

empty

debounce

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/debounce.html

防抖节流，指定时间内的值只接收最新的一个，其他的过滤掉。搜索联想场景适用。

flow {

emit( 1)

delay( 90)

emit( 2)

delay( 90)

emit( 3)

delay( 1010)

emit( 4)

delay( 1010)

emit( 5)

}.debounce( 1000)

// 3 4 5

sample

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/sample.html

采样。给定一个时间周期，仅获取周期内最新发出的值。

flow {

repeat( 10) {

emit(it)

delay( 110)

}

}.sample( 200)

// 13579

//图示

【 1】

| -----------|

1200

2【 3】

| ------------|

200400

distinctUntilChangedBy

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/distinct-until-changed-by.html

去重操作符，判断连续的两个值是否重复，可以选择是否丢弃重复值。

keySelector: (T) -> Any?指定用来判断是否需要比较的 key。

有点类似Recyclerview的DiffUtil机制。

flowOf(

Funny(name = "Tom", age = 8),

Funny(name = "Tom", age = 12),

Funny(name = "Tom", age = 12)

).distinctUntilChangedBy { it.name } .collect { value->

print( value.toString)

}

// Funny(name=Tom, age=8)

distinctUntilChanged

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/distinct-until-changed.html

过滤用，distinctUntilChangedBy的简化调用。连续两个值一样，则跳过发送。

flowOf( 1, 1, 3, 1).distinctUntilChanged

.collect { value->

print( value)

}

// 1 3 1

组合操作符

combine

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/combine.html

组合每个流最新发出的值。

val flow = flowOf( 1, 2).onEach { delay( 10) }

val flow2 = flowOf( "a", "b", "c").onEach { delay( 15) }

flow.combine(flow2) { i, s -> i.toString + s } .collect {

println(it) // Will print "1a 2a 2b 2c"

}

combineTransform

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/combine-transform.html

顾名思义 combine+ transform****

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/combine.html

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform.html

valnumberFlow = flowOf( 1, 2).onEach { delay( 10) }

valstringFlow = flowOf( "a", "b", "c").onEach { delay( 15) }

numberFlow.combineTransform(stringFlow) { number, string ->

emit( " $number: $string" )

}.collect { value ->

println( value )

}

//结果

1:a

2:a

2:b

2:c

merge

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/merge.html

合并多个流成一个流。可以用在多级缓存加载上。

val numberFlow = flowOf( 1, 2).onEach { delay( 10) }

val stringFlow = flowOf( "a", "b", "c").onEach { delay( 15) }

listOf(numberFlow,stringFlow).merge

.collect { value->

print( value)

}

// 1 a 2 b c

flattenConcat

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flatten-concat.html

以顺序方式将给定的流展开为单个流，是Flow<Flow<T>>的扩展函数。

flow {

emit(flowOf( 1, 2, 3))

emit(flowOf( 4, 5, 6))

} .flattenConcat.collect { value->

print( value)

}

// 1 2 3 4 5 6

flattenMerge

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flatten-merge.html

作用和flattenConcat一样，但是可以设置并发收集流的数量。

有个入参：concurrency: Int,当其 == 1时，效果和flattenConcat一样，大于 1 时，则是并发收集。

flow {

emit(flowOf( 1, 2, 3).flowOn(Dispatchers.IO))

emit(flowOf( 4, 5, 6).flowOn(Dispatchers.IO))

emit(flowOf( 7, 8, 9).flowOn(Dispatchers.IO))

}.flattenMerge( 3).collect { value->

print( value)

}

//1 2 3 7 8 9 4 5 6 (顺序并不固定)

flatMapContact

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-concat.html

这是一个组合操作符，相当于 map+ flattenConcat, 通过 map 转成一个流，再通过 flattenConcat。

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map.html

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flatten-concat.html

展开合并成一个流。

flowOf( 1, 2, 3).flatMapConcat {

flowOf(it.toString + " map")

} .collect { value->

print ln( value)

}

// 1 map

// 2 map

// 3 map

flatMapLatest

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/ flat-map-latest.html

和其他带 Latest的操作符一样，如果下个值来了，上变换还没结束，就取消掉。

相当于 transformLatest+ emitAll。

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform-latest.html

flow {

emit( "a")

delay( 100)

emit( "b")

}.flatMapLatest { value->

flow {

emit( value)

delay( 200)

emit( value+ "_last")

}

}.collect { value->

print( value)

}

// a b b_last

flatMapMerge

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-merge.html

也是组合操作符，简化使用。map+ flattenMerge。因此也是有concurrency: Int这样一个参数，来限制并发数。

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map.html

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flatten-merge.html

flowOf( "a", "b", "c", "d", "e", "f").flatMapMerge( 3) { value->

flow {

emit( value)

} .flowOn(Dispatchers.IO)

}.collect { value->

print( value)

}

// b a c d e f

zip

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/zip.html

对两个流进行组合，分别从二者取值，一旦一个流结束了，那整个过程就结束了。

val flow = flowOf( 1, 2, 3).onEach { delay( 10) }

val flow2 = flowOf( "a", "b", "c", "d").onEach { delay( 15) }

flow.zip(flow2) { i, s -> i.toString + s }.collect {

println(it)

}

// Will print "1a 2b 3c"

功能性操作符

cancellable

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/cancellable.html

接收的的时候判断协程是否被取消，如果已取消，则抛出异常。

val job= flowOf( 1, 3, 5, 7).cancellable.onEach { value->

print( value)

} .launchIn(lifecycleScope)

//取消

job.cancel

catch

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/catch.html

对上游异常进行捕获，对下游无影响。

上游指的是此操作符之前的流。

下游指的是此操作符之后的流。

flow< Int> {

throwIOException( "")

} . catch{ e ->

if(e isIOException){

//...

}

retryWhen

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/retry-when.html

有条件的进行重试，lambda 中有两个参数: 一个是异常原因，一个是当前重试的 index (从0开始)。

lambda 的返回值为 Boolean ，true则继续重试 ,false 则结束重试。

flow< Int> {

print( "doing")

throwIOException( "")

} .retryWhen { cause,attempt->

if(attempt > 4){

return@retryWhenfalse

}

cause isIOException

}

retry

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/retry.html

重试机制，当流发生异常时可以重新执行。retryWhen 的简化版。

retries: ``Long`` = Long.MAX_VALUE 指定重试次数，以及控制是否继续重试。(默认为true)

flow< Int> {

throwIOException( "")

}. retry ( 3){ e->

if(e isIOException){

true

} else{

false

}

flow< Int> {

throwIOException( "")

}.retry( 3)

buffer

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/buffer.html

如果操作符的代码需要相当****长时间来执行，可使用buffer操作符在执行期间为其创建一个单独的协程。

capacity: Int = BUFFERED 缓冲区的容量。

onBufferOverflow: BufferOverflow = BufferOverflow.``SUSPEND **溢出的话执行的操作。

有三个选择：SUSPEND 挂起，DROP_OLDEST 丢掉旧的，DROP_LATEST丢掉新的。

flowOf( "A", "B", "C")

.onEach { println( "1$it") }

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

Q : -->-- [ 1A] -- [ 2A] -- [ 1B] -- [ 2B] -- [ 1C] -- [ 2C] -->--

flowOf( "A", "B", "C")

.onEach { println( "1$it") }

.buffer // <--------------- buffer between onEach and collect

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

P : -->-- [ 1A] -- [ 1B] -- [ 1C] ---------->-- // flowOf(...).onEach { ... }

| channel // buffer

Q : -->---------- [ 2A] -- [ 2B] -- [ 2C] -->-- // collect

conflate

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/conflate.html

仅保留最新值, 内部就是 buffer``(``CONFLATED``)

flow {

repeat( 30) {

delay( 100)

emit(it)

}

}.conflate.onEach { delay( 1000) } .collect { value->

print( value)

}

// 0 7 15 22 29 (结果不固定)

flowOn

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow-on.html

指定上游操作的执行线程。想要切换执行线程就用它！

flow. map{ ... } // Will be executed in IO

. flowOn (Dispatchers.IO) // This one takes precedence

. collect{ ... }

总结

以上就是Kotlin Flow所有操作符的基本用法，在实际场景中按需使用。比如上面说的：搜索场景使用debounce防抖，网络请求使用retry,组件通信使用SharedFlow, 数据合并使用combine等操作符。提升开发效率啦噜。

最后推荐一下我做的网站，玩Android: wanandroid.com，包含详尽的知识体系、好用的工具，还有本公众号文章合集，欢迎体验和收藏！

：

Android 12 保姆级适配指南来啦！

Android卡顿掉帧？努比亚技术团队分享

根据手机壁纸自动切换App主题，它真的来了！

点击关注我的公众号

如果你想要跟大家分享你的文章，欢迎投稿~

┏(＾0＾)┛明天见！

源码村长

分享到：

源码村长

1 评论

发表评论

源码村长

相关推荐

1 评论

发表评论