线程切换哪家强?RxJava与Flow的操作符对比

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: Flow作为Coroutine版的RxJava,同RxJava一样可以方便地进行线程切换。 本文针对两者在多线程场景中的使用区别进行一个简单对比。 1. RxJava subscribeOn用来决定在

Flow作为Coroutine版的RxJava,同RxJava一样可以方便地进行线程切换。
本文针对两者在多线程场景中的使用区别进行一个简单对比。

1. RxJava


我们先来回顾一下RxJava中的线程切换
在这里插入图片描述

如上,RxJava使用subscriberOnobserveOn进行线程切换

subscribeOn

subscribeOn用来决定在哪个线程进行订阅,对于Cold流来说即决定了数据的发射线程。使用中有两点注意:

  1. 当调用链上只有一个subscribeOn时,可以出现在任意位置

在这里插入图片描述
在这里插入图片描述

上面两种写法效果是一样的:都是在io线程订阅后发射数据

  1. 当调用链上有多个subscribeOn时,只有第一个生效:

在这里插入图片描述

上面第二个subscribeOn没有意义

observeOn

observeOn用来决定在哪个线程上响应

  1. observeOn决定调用链上下游操作符执行的线程

在这里插入图片描述

上面绿线部分的代码将会运行在主线程

  1. 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订阅的操作符(例如mergezip等),需要留意各自的subscribeOn的线程,以防不符合预期的行为出现。

2. Flow


接下来看一下 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为止

3. Summary


RxJava的Observable与Coroutine的Flow都支持线程切换,相关API的对比如下:

线程池调度 线程操作符 数据源同步创建 异步创建 并发执行
RxJava Schedulers (io(), computation(), mainThread()) subscribeOn, observeOn just deffer{} flatMap(inner subscribeOn)
Flow Dispatchers (IO, Default, Main) flowOn flowOf flow{} flatMapMerge(inner or outer flowOn)

最后通过一个例子看一下如何将代码从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

从日志可以看到,123的时序性以及inner1inner2的并发性与RxJava的一致。

4. FIN


Flow在线程切换方面可以完全取代RxJava的能力,而且将subscribeOnobserveOn两个操作符合二为一成flowOn,学习成本更低。随着flow的操作符种类日趋完善,未来在Android/Kotlin开发中可以跟RxJava说再见了👋🏻

image.png

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
目录
相关文章
|
安全 开发者
容易被忽略的知识点:RxJava操作符的线程安全
RxJava操作符大多不是线程安全的,如果所以写代码时不注意这一点很容易发生bug。本文将分享如何写成更安全的Rxjava代码
264 0
容易被忽略的知识点:RxJava操作符的线程安全
|
1月前
|
NoSQL Redis
单线程传奇Redis,为何引入多线程?
Redis 4.0 引入多线程支持,主要用于后台对象删除、处理阻塞命令和网络 I/O 等操作,以提高并发性和性能。尽管如此,Redis 仍保留单线程执行模型处理客户端请求,确保高效性和简单性。多线程仅用于优化后台任务,如异步删除过期对象和分担读写操作,从而提升整体性能。
68 1
|
3月前
|
存储 消息中间件 资源调度
C++ 多线程之初识多线程
这篇文章介绍了C++多线程的基本概念,包括进程和线程的定义、并发的实现方式,以及如何在C++中创建和管理线程,包括使用`std::thread`库、线程的join和detach方法,并通过示例代码展示了如何创建和使用多线程。
72 1
|
3月前
|
Java 开发者
在Java多线程编程中,创建线程的方法有两种:继承Thread类和实现Runnable接口
【10月更文挑战第20天】在Java多线程编程中,创建线程的方法有两种:继承Thread类和实现Runnable接口。本文揭示了这两种方式的微妙差异和潜在陷阱,帮助你更好地理解和选择适合项目需求的线程创建方式。
53 3
|
3月前
|
Java 开发者
在Java多线程编程中,选择合适的线程创建方法至关重要
【10月更文挑战第20天】在Java多线程编程中,选择合适的线程创建方法至关重要。本文通过案例分析,探讨了继承Thread类和实现Runnable接口两种方法的优缺点及适用场景,帮助开发者做出明智的选择。
35 2
|
3月前
|
Java
Java中多线程编程的基本概念和创建线程的两种主要方式:继承Thread类和实现Runnable接口
【10月更文挑战第20天】《JAVA多线程深度解析:线程的创建之路》介绍了Java中多线程编程的基本概念和创建线程的两种主要方式:继承Thread类和实现Runnable接口。文章详细讲解了每种方式的实现方法、优缺点及适用场景,帮助读者更好地理解和掌握多线程编程技术,为复杂任务的高效处理奠定基础。
56 2
|
3月前
|
Java 开发者
Java多线程初学者指南:介绍通过继承Thread类与实现Runnable接口两种方式创建线程的方法及其优缺点
【10月更文挑战第20天】Java多线程初学者指南:介绍通过继承Thread类与实现Runnable接口两种方式创建线程的方法及其优缺点,重点解析为何实现Runnable接口更具灵活性、资源共享及易于管理的优势。
58 1
|
3月前
|
安全 Java 开发者
Java多线程中的`wait()`、`notify()`和`notifyAll()`方法,探讨了它们在实现线程间通信和同步中的关键作用
本文深入解析了Java多线程中的`wait()`、`notify()`和`notifyAll()`方法,探讨了它们在实现线程间通信和同步中的关键作用。通过示例代码展示了如何正确使用这些方法,并分享了最佳实践,帮助开发者避免常见陷阱,提高多线程程序的稳定性和效率。
71 1
|
3月前
|
Java
在Java多线程编程中,`wait()` 和 `notify()/notifyAll()` 方法是线程间通信的核心机制。
在Java多线程编程中,`wait()` 和 `notify()/notifyAll()` 方法是线程间通信的核心机制。它们通过基于锁的方式,使线程在条件不满足时进入休眠状态,并在条件成立时被唤醒,从而有效解决数据一致性和同步问题。本文通过对比其他通信机制,展示了 `wait()` 和 `notify()` 的优势,并通过生产者-消费者模型的示例代码,详细说明了其使用方法和重要性。
59 1