我们先来回顾一下RxJava中的线程切换。
如上,RxJava使用subscriberOn与observeOn进行线程切换。
subscribeOn
subscribeOn用来决定在哪个线程进行订阅,对于Cold流来说即决定了数据的发射线程。使用中有两点注意:
1、当调用链上只有一个subscribeOn时,可以出现在任意位置。
上面两种写法效果是一样的:都是在io线程订阅后发射数据。
2、当调用链上有多个subscribeOn时,只有第一个生效:
上面第二个subscribeOn没有实际意义。
observeOn
observeOn用来决定在哪个线程上响应:
1、observeOn决定调用链上下游操作符执行的线程。
上面绿线部分的代码将会运行在主线程。
2、与subscribeOn不同,调用链上允许存在多个observeOn且每个都有效。
上面蓝色绿色部分因为observeOn的存在分别切换到了不同线程执行。
just
RxJava的初学者经常会犯的一个错误是在Observable.just(...)里做耗时任务。just并不是接受lambda,所以是立即执行的,不受subscribeOn的影响。
如上,loadDataSync()不会在io执行。
想要在io执行,需要使用Observable.deffer{}。
flatMap
结合上面介绍的RxJava的线程切换,看下面这段代码。
如果我们希望loadData(id)并发执行,那么上面的写法是错误的。
subscribe(io())意味着其上游的数据在单一线程中串行发射。因此虽然flatMap{}返回多个Observable, 都是都在单一线程中订阅,多个loadData始终运行在同一线程。
代码经过一下修改后,可以达到并发执行的效果:
当订阅flatMap返回的Observable时,通过subscribeOn分别指定订阅线程。
其他类似flatMap这种涉及多个Observable订阅的操作符(例如merge、zip等),需要留意各自的subscribeOn的线程,以防不符合预期的行为出现。
接下来看一下 Flow的线程切换 。
Flow是基于CoroutineContext进行线程切换,所以这部分内容需要你对Croutine事先有基本的了解。
flowOn类似于RxJava的subscribeOn,Flow中没有对应observeOn的操作符,因为collect是一个suspend函数,必须在CoroutineScope中执行,所以响应线程是由CoroutineContext决定的。例如你在main中执行collect,那么响应线程就是Dispatcher.Main。
flowOn
说flowOn类似于subscribeOn,因为它们都可以用来决定上游线程 上面代码中,flowOn前面代码将会在IO执行。
与subscribeOn不同的是,flowOn允许出现多次,每个都会影响其前面的操作。
上面代码,根据颜色可以看出来flowOn影响的范围。
launchIn
collect是suspend函数,所以后续代码因为协程挂起不会继续执行。
所以上面代码可能会不符合预期,因为第一个collect不走完第二个走不到。
正确的写法是为每个collect单独起一个协程。
或者使用launchIn,写法更加优雅 。
launchIn不会挂起协程,所以与RxJava的subscribe更加接近。
通过名字可以感觉出来launchIn只不过是之前例子中launch的一个链式调用的语法糖。
flowOf
flowOf类似于Observable.just(),需要注意flowOf内的内容是立即执行的,不受flowOn影响。
希望calculate()运行在IO,可以使用flow{ }。
flatMapMerge
flatMapMerge类似RxJava的flatMap 。
如上,2个item各自flatMap成2个item,即一共发射了4条数据,日志输出如下:
inner: pool-2-thread-2 @coroutine#4
inner: pool-2-thread-3 @coroutine#5
inner: pool-2-thread-3 @coroutine#5
inner: pool-2-thread-2 @coroutine#4
collect: pool-1-thread-2 @coroutine#2
collect: pool-1-thread-2 @coroutine#2
collect: pool-1-thread-2 @coroutine#2
collect: pool-1-thread-2 @coroutine#2
通过日志我们发现flowOn虽然写在flatMapMerge外面,inner的日志却可以打印在多个线程上(都来自pool2线程池),这与flatMap是不同的,同样场景下flatMap只能运行在线程池的固定线程上。
如果将flowOn写在flatMapMerge内部。
结果如下:
inner: pool-2-thread-2 @coroutine#6
inner: pool-2-thread-1 @coroutine#7
inner: pool-2-thread-2 @coroutine#6
inner: pool-2-thread-1 @coroutine#7
collect: pool-1-thread-3 @coroutine#2
collect: pool-1-thread-3 @coroutine#2
collect: pool-1-thread-3 @coroutine#2
collect: pool-1-thread-3 @coroutine#2
inner仍然打印在多个线程,flowOn无论写在flatMapMerge内部还是外部,对flatMapMerge内的处理没有区别。
但是flatMapMerge之外还是有区别的,看下面两段代码 。
通过颜色可以知道flowOn影响的范围,向上追溯到flowOf为止。
RxJava的Observable与Coroutine的Flow都支持线程切换,相关API的对比如下:
最后通过一个例子看一下如何将代码从RxJava迁移到Flow。
RxJava
RxJava代码如下:
使用到的Schedulers定义如下:
代码执行结果:
1: pool-1-thread-1
1: pool-1-thread-1
1: pool-1-thread-1
2: pool-3-thread-1
2: pool-3-thread-1
2: pool-3-thread-1
inner 1: pool-4-thread-1
inner 1: pool-4-thread-2
inner 1: pool-4-thread-1
inner 1: pool-4-thread-1
inner 1: pool-4-thread-2
inner 1: pool-4-thread-2
inner 1: pool-4-thread-3
inner 2: pool-5-thread-1
inner 2: pool-5-thread-2
3: pool-5-thread-1
inner 2: pool-5-thread-2
inner 1: pool-4-thread-3
inner 2: pool-5-thread-2
inner 2: pool-5-thread-3
3: pool-5-thread-1
3: pool-5-thread-1
3: pool-5-thread-1
end: pool-6-thread-1
end: pool-6-thread-1
inner 1: pool-4-thread-3
end: pool-6-thread-1
3: pool-5-thread-1
inner 2: pool-5-thread-1
3: pool-5-thread-1
inner 2: pool-5-thread-3
inner 2: pool-5-thread-1
end: pool-6-thread-1
3: pool-5-thread-3
3: pool-5-thread-3
end: pool-6-thread-1
inner 2: pool-5-thread-3
3: pool-5-thread-3
end: pool-6-thread-1
end: pool-6-thread-1
end: pool-6-thread-1
end: pool-6-thread-1
代码较长,通过颜色标记法帮我们理清线程关系。
上色后一目了然了,需要特别注意的是由于flatMap中切换了数据源的同时切换了线程,所以打印 3的线程不是s2 而是 s4
Flow
首先创建对应的Dispatcher。
然后将代码换成Flow的写法,主要遵循下列原则。
RxJava通过observeOn切换后续代码的线程。
Flow通过flowOn切换前置代码的线程。
打印结果如下:
1: pool-1-thread-1 @coroutine#6
1: pool-1-thread-1 @coroutine#6
1: pool-1-thread-1 @coroutine#6
2: pool-2-thread-2 @coroutine#5
2: pool-2-thread-2 @coroutine#5
2: pool-2-thread-2 @coroutine#5
inner 1: pool-3-thread-1 @coroutine#10
inner 1: pool-3-thread-2 @coroutine#11
inner 1: pool-3-thread-3 @coroutine#12
inner 1: pool-3-thread-2 @coroutine#11
inner 1: pool-3-thread-3 @coroutine#12
inner 2: pool-4-thread-3 @coroutine#9
inner 1: pool-3-thread-1 @coroutine#10
inner 1: pool-3-thread-3 @coroutine#12
inner 1: pool-3-thread-2 @coroutine#11
inner 2: pool-4-thread-1 @coroutine#7
inner 2: pool-4-thread-2 @coroutine#8
inner 2: pool-4-thread-1 @coroutine#7
inner 2: pool-4-thread-3 @coroutine#9
inner 1: pool-3-thread-1 @coroutine#10
3: pool-4-thread-1 @coroutine#3
inner 2: pool-4-thread-3 @coroutine#9
inner 2: pool-4-thread-2 @coroutine#8
end: pool-5-thread-1 @coroutine#2
3: pool-4-thread-1 @coroutine#3
inner 2: pool-4-thread-2 @coroutine#8
3: pool-4-thread-1 @coroutine#3
end: pool-5-thread-1 @coroutine#2
3: pool-4-thread-1 @coroutine#3
end: pool-5-thread-1 @coroutine#2
end: pool-5-thread-1 @coroutine#2
3: pool-4-thread-1 @coroutine#3
3: pool-4-thread-1 @coroutine#3
end: pool-5-thread-1 @coroutine#2
end: pool-5-thread-1 @coroutine#2
3: pool-4-thread-1 @coroutine#3
3: pool-4-thread-1 @coroutine#3
end: pool-5-thread-1 @coroutine#2
end: pool-5-thread-1 @coroutine#2
inner 2: pool-4-thread-1 @coroutine#7
3: pool-4-thread-1 @coroutine#3
end: pool-5-thread-1 @coroutine#2
从日志可以看到,1、2、3的时序性以及inner1和inner2的并发性与RxJava的一致。
Flow在线程切换方面可以完全取代RxJava的能力,而且将subscribeOn和observeOn两个操作符合二为一成flowOn,学习成本更低。
随着flow的操作符种类日趋完善,未来在Android/Kotlin开发中可以跟RxJava说再见了。