认识达内从这里开始

我们在学习java开发编程技术的时候，相信大家都学习过关于多线程编程方式。今天，我们就一起来了解一下多线程编程方法都有哪些优势帮助我们提高软件开发效率。

为何要进行反应式编程?

任何一篇关于RxJava的文章都会包含一个“为什么要进行反应式编程”的章节，我们也不打算破坏这个约定。在Android中采用反应式的方式构建App会带来多项收益，接下来我们讨论几项你真正值得关注的好处。

不再会有回调地狱

如果你已经做过一段时间的Android开发的话，你肯定会明白嵌套回调会快速地让所有事情失去控制。

如果你想按顺序执行一系列的异步操作并且下一步操作的行为要依赖于上一步操作的结果，那么你就会遇到这种状况。几乎瞬间，代码就会变得超级丑陋和难以管理。

简单的错误处理

在命令编程的世界中，如果你要执行大量复杂、异步的操作，错误可能会在任意的地方出现，为了处理这些场景，我们就需要将大量补丁式的代码到处放得到处都是，从而形成大量重复和繁琐的代码。

超级简单的多线程

我们都知道(私下也承认)在Java中编写多线程代码有多困难。在后台线程中执行一段代码并在UI线程中获取结果，这听起来似乎很简单，但实际上有很多复杂的情况需要处理。

通过使用RxJava，你可以非常容易地在任意线程中执行复杂的操作，它会维持适当的状态同步，并能够让你无缝地切换线程。

RxJava所带来的好处是无穷无尽的，就此可以谈论几个小时，但是现在我们要更深入地探索一下它为多线程编程所带来的真正威力。

默认情况下，RxJava并不是多线程的

是的，你没有看错。RxJava默认情况下跟多线程一点关系都没有。在官方网站上，RxJava是这样定义的：

借助observable序列，在Java VM上组合异步和基于事件的程序的库。

看到“异步”这个词，很多人就形成了一种误解，认为RxJava默认就是多线程的。的确，它支持多线程并且提供了一些强大的特性以便于我们执行异步操作，但是不能就此断定它的默认行为就是多线程的。

如果你多少了解一些RxJava的话，会知道它的基本构造为：

源Observable

一个或多个操作符(Operator)

目标Subscriber

使用Scheduler

RxJava中的多线程操作主要是由强大的Scheduler集合提供的。在RxJava中，我们无法直接访问或操作线程。如果想要使用线程的话，必须要通过内置的Scheduler来实现。

你可以将Scheduler视为线程或线程池(一个线程的集合)，能够执行不同类型的工作。

简而言之，如果你需要在特定的线程中执行任务的话，我们就需要此选择恰当的Scheduler，Scheduler接下来会从它的池中获取一个可用的线程，并基于该线程执行任务。

在RxJava框架中有多种类型的Scheduler，但是这里比较有技巧的一点就是为合适的工作选择恰当的Scheduler。如果你没有选择恰当的Scheduler的话，那么任务就无法最优地运行，所以接下来，我们尝试理解每一个Scheduler。

Schedulers.io()

这是由无边界线程池作为支撑的一个Scheduler，它适用于非CPU密集的I/O工作，比如访问文件系统、执行网络调用、访问数据库等等。这个Scheduler是没有限制的，它的线程池可以按需一直增长。

Schedulers.computation()

这个Scheduler用于执行CPU密集的工作，比如处理大规模的数据集、图像处理等等。它由一个有界的线程池作为支撑，线程的最大数量就是可用的处理器数量。

因为这个Scheduler只适用于CPU密集的任务，我们希望限制线程的数量，这样的话，它们不会彼此抢占CPU时间或出现线程饿死的现象。

Schedulers.newThread()

这个Scheduler 每次都会创建一个全新的线程来完成一组工作。它不会从任何线程池中受益，线程的创建和销毁都是很昂贵的，所以你需要非常小心，不要衍生出太多的线程，导致服务器系统变慢或出现内存溢出的错误。

理想情况下，你应该很少使用这个Scheduler，它大多用于在一个完全分离的线程中开始一项长时间运行、隔离的一组任务。

Schedulers.single()

这个Scheduler是RxJava 2新引入的，它的背后只有一个线程作为支撑，只能按照有序的方式执行任务。如果你有一组后台任务要在App的不同地方执行，但是同时只能承受一个任务执行的话，那么这个Scheduler就可以派上用场了。

Schedulers.from(Executor executor)

我们可以使用它创建自定义的Scheduler，它是由我们自己的Executor作为支撑的。在有些场景下，我们希望创建自定义的Scheduler为App执行特定的任务，这些任务可能需要自定义的线程逻辑。

假设，我们想要限制App中并行网络请求的数量，那么我们就可以创建一个自定义的Scheduler，使其具有一个固定线程池大小的Executor：Scheduler.from(Executors.newFixedThreadPool(n))，然后将其应用到代码中所有网络相关的Observable上。

AndroidSchedulers.mainThread()

这是一个特殊的Scheduler，它无法在核心RxJava库中使用，要使用它，必须要借助RxAndroid扩展库。这个Scheduler对Android App特别有用，它能够在应用的主线程中执行基于UI的任务。

默认情况下，它会在应用主线程关联的looper中进行任务排队，但是它有一个其他的变种，允许我们以API的形式使用任意的Looper：AndroidSchedulers.from(Looper looper)。

注意：在使用无边界线程池支撑的Scheduler时，比如Schedulers.io()，我们要特别小心，因为它有可能会导致线程池无限增长，使系统中出现大量的线程。

理解subscribeOn()和observeOn()

现在，我们对不同类型的Scheduler已经有了一个清晰的理解，并且掌握了何时该使用哪种Scheduler，我们继续前进，接下来会详细介绍subscribeOn()和observeOn()操作符。

要理解RxJava真正的多线程功能，我们需要深入理解这两个操作符分别如何运行以及何时要将它们组合起来。

subscribeOn()

简而言之，这个操作符指定源observable要在哪个线程中发出各个条目。读者需要理解这里“源(source)” observable的含义，如果你有一个observable链的话，源observable始终位于根部或者说是链的顶部，也就是产出物(emission)生成的地方。

读者也看到了，如果我们不使用subscribeOn()的话，所有的产出都是直接在代码执行的线程中生成的(在我们的场景中，也就是main线程)。

接下来，我们将所有的产出物在计算线程中生成，这需要使用subscribeOn()和 Schedulers.computation() Scheduler。如果你运行下面的代码片段，所有产出物是在线程池中某个可用的计算线程中生成的，RxComputationThreadPool-1。

简洁起见，我们没有使用完整的DisposableSubscriber，因为在这种简单的场景中，我们不需要每次都处理onError()和onComplete()，我们只需要处理onNext()，它只是一个简单的消费者。

背后的原理

你可能已经猜到，为何是最后的observeOn()，而不是其他的observeOn()发挥作用了。我们已经知道，订阅只能针对上游才能发生，另外一方面，产出物只能面对下游才能发挥作用。它们从源observable开始，一路沿着链向下，直至最后的sink subscriber。

observeOn()操作符只能针对下游observable才有效，所以最后的observeOn()和Schedulers.computation()覆盖了前面声明的observeOn()操作符。因此，每当你想为特定的observable切换线程时，需要做的就是指定observeOn()操作符。同步、状态不一致性、竞态条件以及其他线程相关的问题都会在幕后自动处理好了。

到目前为止，相信你已经对如何使用RxJava编写多线程的应用并保证App的快速和流畅有了很好的理解。

作者：Aritra Roy

译者：张卫滨

来源：infoq

【免责声明】：本内容转载于网络，转载目的在于传递最新信息。文章内容为作者个人意见，本平台对文中陈述、观点保持中立，不对所包含内容的准确性、可靠性与完整性提供形式地保证。请读者仅作参考。