- 1. 主要导包
- 2. map 转换 *** 作符
- 3. filter过滤 *** 作符
- 4. take限长 *** 作符
- 5. drop丢弃 *** 作符
- 6. flowOn *** 作符
- 7. onStart *** 作符
- 8. onCompletion完成 *** 作符
- 9. catch异常收集 *** 作符
- 10. buffer缓冲 *** 作符
- 11. conflate 合并 *** 作符
- 参考
//协程包
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.1"
//主要导入viewModelScope，lifecycleScope作用域，在View界面 *** 作协程能保证界面安全
implementation 'androidx.lifecycle:lifecycle-livedata-ktx:2.3.1'
implementation "androidx.lifecycle:lifecycle-viewmodel-ktx:2.3.1"
implementation "androidx.lifecycle:lifecycle-runtime-ktx:2.3.1"
implementation "androidx.lifecycle:lifecycle-extensions:2.2.0"
//工具包
implementation "com.blankj:utilcodex:1.29.0"
map *** 作符可以看成Rxjava中的map *** 作符
//创建一个流，每隔一秒发送一个Int数据
private fun simple(): Flow<Int> = flow {
for (i in 1..3) {
delay(1000)
LogUtils.d("Emitting $i")
emit(i)
}
}
//viewModel中调用，将Int转成String
viewModelScope.launch {
simple()
.map { "response $it" }
.collect {//数据接收
LogUtils.d("result $it")
}
}
过滤掉自己想要的数据
private fun simple(): Flow<Int> = flow {
for (i in 1..3) {
delay(1000)
LogUtils.d("Emitting $i")
emit(i)
}
}
//viewModel中调用filter *** 作符
fun flowFilter() {
viewModelScope.launch {
simple()
.filter { it != 2 }//过滤掉不为2的数据
.collect {
LogUtils.d("result $it")
}
}
}
结果如下：
D/Collect: result 1
D/Collect: result 3
限长 *** 作符,过滤型 *** 作符，与Rxjava中的take相似,只输出输入数字的前面元素，比如take值是2,则只输出前面两个数
private fun simple(): Flow<Int> = flow {
for (i in 1..5) {
delay(1000)
LogUtils.d("Emitting $i")
emit(i)
}
}
//take *** 作符
fun flowTake() {
viewModelScope.launch {
simple()
.take(2)//只获取前面两个数据
.collect {
LogUtils.d("result $it")
}
}
}
D/Collect: result 1
D/Collect: result 2
private fun simple(): Flow<Int> = flow {
for (i in 1..3) {
delay(1000)
LogUtils.d("Emitting $i")
emit(i)
}
}
fun flowDrop() {
viewModelScope.launch {
simple()
.drop(2)//丢到前面两个数据
.collect() {
LogUtils.d("result $it")
}
}
}
D/Collect: result 3
flowOn 函数，该函数用于更改流发射的上下文，即作用域。简单的来说就是切换线程。flowOn之前的函数线程在flowOn指定的线程作用域，flowOn之后的函数在launch的线程作用域。
private fun flowOn() {
//launch：流在主线程启动，flowOn之后的函数collect在主线程中
//viewModelScope默认作用域就是主线程，这里可以不用切
viewModelScope.launch(Dispatchers.Main) {
flow {
for (i in 1..3) {
delay(1000)
LogUtils.d("Thread is ${Thread.currentThread().name} Emitting $i")
emit(i)
}
}.flowOn(Dispatchers.Default)// //flowOn切换到默认作用域，flow{}中的方法执行在默认线程中，子线程中
.collect {
LogUtils.d("Thread is ${Thread.currentThread().name} collect $it")
}
}
}
Thread is DefaultDispatcher-worker-1 Emitting 1
Thread is main collect 1
Thread is DefaultDispatcher-worker-1 Emitting 2
Thread is main collect 2
Thread is DefaultDispatcher-worker-1 Emitting 3
Thread is main collect 3
流执行开始之前调用
private fun flowStart() {
viewModelScope.launch(Dispatchers.Main) {
flow {
for (i in 1..3) {
delay(1000)
LogUtils.d("Thread is ${Thread.currentThread().name} Emitting $i")
emit(i)
}
}.flowOn(Dispatchers.Default)
.onStart { //flow流执行开始之前
}.collect {
LogUtils.d("Thread is ${Thread.currentThread().name} collect $it")
}
}
}
流执行完成后调用
private fun flowCompletion() {
viewModelScope.launch(Dispatchers.Main) {
flow {
for (i in 1..3) {
delay(1000)
LogUtils.d("Thread is ${Thread.currentThread().name} Emitting $i")
emit(i)
}
}.flowOn(Dispatchers.Default)
.onCompletion { //flow流执行开始之前
}.collect {
LogUtils.d("Thread is ${Thread.currentThread().name} collect $it")
}
}
}
fun flowCatch() {
viewModelScope.launch(Dispatchers.Main) {
flow {
for (i in 1..3) {
delay(1000)
if (i == 2) {
throw RuntimeException("Error")//数据为2时候报异常
}
emit(i)
}
}.flowOn(Dispatchers.Default)
.catch { error ->
LogUtils.d("Thread is ${Thread.currentThread().name} error $error")
}.collect {
LogUtils.d("Thread is ${Thread.currentThread().name} collect $it")
}
}
}
D/Collect: Thread is main collect 1
D/FlowViewModel: Thread is main error java.lang.RuntimeException: Error
从收集流所花费的时间来看，将流的不同部分运行在不同的协程中将会很有帮助，特别是当涉及到长时间运行的异步 *** 作时。例如，考虑一种情况， 一个 simpleBuffer 流的发射很慢，它每花费 100 毫秒才产生一个元素；而收集器也非常慢， 需要花费 300 毫秒来处理元素。
收发速度不一致时。
//发送数据过快，处理数据过慢
private fun simpleBuffer(): Flow<Int> = flow {
for (i in 1..3) {
delay(100)
LogUtils.d(" Emitting $i")
emit(i)
}
}
fun flowBuffer(){
viewModelScope.launch(Dispatchers.Main) {
simpleBuffer()
.collect {
delay(300)
LogUtils.d(" collect $it")
}
}
}
不加buffer() *** 作符，执行时间大约在1113ms，添加buffer *** 作后,执行时间为910ms
fun flowBuffer(){
viewModelScope.launch(Dispatchers.Main) {
simpleBuffer()
.buffer().collect {
delay(300)
LogUtils.d(" collect $it")
}
}
}
//加载buffer *** 作符的情况
11:20:22.615 13265-13265/com.hjq.viewbinding D/FlowViewModel: Emitting 1
11:20:22.717 13265-13265/com.hjq.viewbinding D/FlowViewModel: Emitting 2
11:20:22.819 13265-13265/com.hjq.viewbinding D/FlowViewModel: Emitting 3
11:20:22.919 13265-13265/com.hjq.viewbinding D/Collect: collect 1
11:20:23.222 13265-13265/com.hjq.viewbinding D/Collect: collect 2
11:20:23.527 13265-13265/com.hjq.viewbinding D/Collect: collect 3
当流代表部分 *** 作结果或 *** 作状态更新时，可能没有必要处理每个值，而是只处理最新的那个。在本示例中，当收集器处理它们太慢的时候， conflate *** 作符可以用于跳过中间值。
fun flowConflate() {
viewModelScope.launch {
flow {
for (i in 1..3) {
delay(100)
LogUtils.d(" Emitting $i")
emit(i)
}
}.conflate().collect {
delay(300)
LogUtils.d(" collect $it")
}
}
}
//接收时，2数据丢了
D/FlowViewModel: Emitting 1
D/FlowViewModel: Emitting 2
D/FlowViewModel: Emitting 3
D/Collect: collect 1
D/Collect: collect 3
协程中文网
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）