作者：沈哲
出版社：电子工业出版社
格式: AZW3, DOCX, EPUB, MOBI, PDF, TXT
RxJava 2.x实战试读：
前言
编写这本书的目的
笔者从2015年开始关注RxJava以及响应式编程，当时网上的资料很少。三年后的今天，我们可以看到越来越多的App都在使用Rx相关的技术。在Java后端，Spring 5也开始支持响应式编程。在开源的技术社区里，Rx、响应式编程、函数式编程都是热门话题。我们公司开发的App，笔者也会强制大家使用RxJava或者RxSwfit。本书通过完整的体系介绍了RxJava的方方面面。
对于不了解响应式编程的开发者，RxJava的入门可能会有一些难度。笔者结合自身的学习以及使用情况，尽可能使用通俗易懂的讲解方式带领大家学习RxJava。同时，本书还附有丰富的例子，从Android开发到Java后端的开发，相信能够让大家感受到RxJava的魅力。
读者对象
1）Android开发工程师。
2）Java后端开发工程师。
3）对函数式编程、响应式编程感兴趣的IT从业人员。
如何阅读本书
本书共分为18章。
第1章讲述了RxJava的来龙去脉，以及什么是响应式编程、什么是函数式编程。
第2章至第7章讲述了RxJava的基础概念，以及RxJava常用的操作符。
第8章至第10章为RxJava的高级部分。
第11章至第12章为RxAndroid的使用篇，介绍了常用的RxBinding、Retrofit等框架。
第13章至第15章为RxJava的实战篇，介绍了如何实现一个基于RxJava的Event Bus框架，以及Spring Boot如何与RxJava相结合使用。
第16章至第18章介绍了Java 8函数式编程的特性和Kotlin，并展望未来。
勘误和支持
本书相关例子的源码都在 GitHub 上，地址：https：//github.com/fengzhizi715/RxJavaInAction。
由于笔者水平有限，编写本书时难免会出现错误或者纰漏，恳请读者批评指正。读者可以关注笔者的公众号与笔者进行互动。或者通过邮箱：fengzhizi715@126.com，有关本书的任何问题都可以反馈给笔者，笔者期待与您的技术交流。
致谢
首先要感谢我的家人，最主要是感谢我的妻子。在写书期间，恰逢儿子的出生，她承担了绝大部分照顾儿子的责任。
感谢公司的支持与同事的帮助，特别是刘志强帮我整理了很多RxJava相关的资料，以及对本书部分章节进行了试读，并提出意见。
感谢www.bsfans.com罗波同学提供UI支持。 第1章 RxJava简介1.1 你需要了解的函数响应式编程
如果你曾经使用过 Java，那么你一定听说过面向对象的编程思想，也可能听说过 AOP （Aspect Orient Programming，面向切面编程）的编程思想。本书要讲的是全新的编程思想，跟它们没有丝毫的联系。
1.响应式编程（Reactive Programming，简称RP）
在计算机中，响应式编程是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。
传统的编程方式是顺序执行的，需要等待直至完成上一个任务之后才会执行下一个任务。无论是提升机器的性能还是代码的性能，本质上都需要依赖上一个任务的完成。如果需要响应迅速，就得把同步执行的方式换成异步执行，方法执行变成消息发送。这就是异步编程的方式，它是响应式编程的重要特性之一。
响应式编程有以下几个特点。
◎ 异步编程：提供了合适的异步编程模型，能够挖掘多核 CPU 的能力、提高效率、降低延迟和阻塞等。
◎ 数据流：基于数据流模型，响应式编程提供一套统一的Stream风格的数据处理接口。与Java 8中的Stream相比，响应式编程除了支持静态数据流，还支持动态数据流，并且允许复用和同时接入多个订阅者。
◎ 变化传播：简单来说就是以一个数据流为输入，经过一连串操作转化为另一个数据流，然后分发给各个订阅者的过程。这就有点像函数式编程中的组合函数，将多个函数串联起来，把一组输入数据转化为格式迥异的输出数据。
响应式编程一方面在用户界面编程领域及基于实时系统的动画方面都有广泛的应用；另一方面，在处理嵌套回调的异步事件、复杂的列表过滤和变换的时候也都有良好的表现。
现在的App无论是原生H5、HTML5还是Hybird，都会和与数据事件相关的 UI 事件进行大量的交互，使用响应式编程会显得更加得心应手。
这些年来前端比较流行的响应式设计，实际上是指网页能够自动调整布局和样式以适配不同尺寸的屏幕，与我们谈论的响应式编程是两个完全不同的概念。
2.函数式编程（Functional Programming，简称FP）
随着硬件能力的不断提升，单核 CPU 的计算能力几乎达到极限，CPU 已经进入了多核时代，程序员转而通过并发编程、分布式系统来应对越来越复杂的计算任务。
然而并发编程并不是银弹，作为一种基于共享内存的并发编程，多线程编程有常见的死锁、线程饥饿、竞争条件等问题，而且多线程的Bug也难以重现和定位。于是，函数式编程开始兴起。
在函数式编程中，由于数据是不可变的（immutable），因此没有并发编程的问题，是线程安全的。它将计算机运算看作数学中函数的计算，主要特点是将计算过程分解成多个可复用的函数，并且避免了状态及变量的概念。函数式编程虽然也可以归结到面向过程的程序设计，但其思想更接近数学计算。
函数式编程具有以下特点。
◎ 函数是“第一等公民”：所谓“第一等公民”（First Class），指的是函数与其他数据类型一样，处于平等地位，可以赋值给其他变量，也可以作为参数，传入另一个函数，或者作为别的函数的返回值。
◎ 闭包和高阶函数：闭包是起函数的作用并可以像对象一样操作的对象。与此类似，FP语言支持高阶函数。高阶函数可以用另一个函数（间接地，用一个表达式）作为输入参数，在某些情况下，它甚至返回一个函数作为输出参数。这两种结构结合在一起，即可以用优雅的方式进行模块化编程，这是使用FP的最大好处。
◎ 递归：把递归作为控制流程的机制。例如，在 Haskell 的世界中，没有变量赋值和流程跳转，如果要实现一些简单的功能，比如求一个数组中的最大值，则需要借助递归来实现。
◎ 惰性求值（Lazy Evaluation）：它表示为“延迟求值”和“最小化求值”。惰性求值使得代码具备了巨大的优化潜能。支持惰性求值的编译器会像数学家看待代数表达式那样看待函数式编程的程序，即抵消相同项从而避免执行无谓的代码，安排代码执行顺序从而实现更高的执行效率甚至是减少错误。惰性求值另一个重要的好处是它可以构造一个无限的数据类型，无须担心由无穷计算所导致的内存溢出错误。
◎ 没有“副作用”（Side Effect）：指的是函数内部与外部互动（最典型的情况，就是修改全局变量的值），产生运算以外的其他结果。函数式编程强调没有“副作用”，意味着函数要保持独立，所有功能就是返回一个新的值，没有其他行为，尤其是不得修改外部变量的值。
既然函数式编程已经能够解决并发的问题，那为何还需要响应式编程呢？
3.函数响应式编程（Functional Reactive Programming，简称FRP）
函数响应式编程结合了函数式和响应式的优点，把函数范式里的一套思路和响应式编程合起来就是函数响应式编程。
我们知道，传统的面向对象编程是通过抽象出的对象关系来解决问题，函数式编程是通过函数（function）的组合来解决问题，响应式编程是通过函数式编程的方式来解决回调地狱（Callback Hell）的问题。
用传统的面向对象来处理异步事件不是很直观，处理并发也十分麻烦，所以才产生了函数响应式编程。
下面一起进入RxJava的世界，体验一下函数响应式编程带来的乐趣。1.2 RxJava简介
1.RxJava产生的由来
RxJava是Reactive Extensions的Java实现，用于通过使用Observable/Flowable序列来构建异步和基于事件的程序的库。
RxJava扩展观察者模式以支持数据/事件序列，并添加允许你以声明方式组合序列的操作符，同时提取对低优先级的线程、同步、线程安全性和并发数据结构等问题的隐藏。
2.什么是Rx
ReactiveX是Reactive Extensions的缩写，一般简写为Rx，最初是LINQ的一个扩展，由微软架构师Erik Meijer领导的团队所开发，于2012年11月开源。Rx是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便地处理异步数据流。Rx库支持.NET、JavaScript和C++。由于Rx近几年越来越流行，因此现在已经支持几乎全部的流行编程语言。Rx的大部分语言库由 ReactiveX 这个组织负责维护，比较流行的有 RxJava、RxJS 和 Rx.NET，社区网站是http：//reactivex.io/。
3.ReactiveX的历史
微软给的定义是，Rx是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序，通过使用Rx，开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流，用Schedulers参数化异步数据流的并发处理。可以这样定义Rx：Rx=Observables+LINQ+Schedulers。
注：在RxJava 2.x中，Observables包含了Observables、Flowable、Single、Maybe、Completable等。
ReactiveX.io给的定义是，Rx是一个使用可观察数据流进行异步编程的编程接口，ReactiveX结合了观察者模式、迭代器模式和函数式编程的精华。
这个定义听上去很拗口，不过没有关系，在第 2 章和第 4 章会分别讲述 Observable 和Scheduler。
4.Rx模式（1）使用观察者模式。
◎ 创建：Rx可以方便地创建事件流和数据流。
◎ 组合：Rx使用查询式的操作符组合和变换数据流。
◎ 监听：Rx可以订阅任何可观察的数据流并执行操作。（2）简化代码。
◎ 函数式风格：对可观察的数据流使用无副作用的输入/输出函数，避免了程序里错综复杂的状态。
◎ 简化代码：Rx的操作符通常可以将复杂的难题简化为很少的几行代码。
◎ 异步错误处理：传统的 try/catch 没办法处理异步计算，Rx 提供了合适的错误处理机制。
◎ 轻松使用并发：Rx的Observables（包括Observable、Flowable、Single、Completable和Maybe）和Schedulers可以让开发者摆脱底层的线程同步和各种并发问题。
总而言之，RxJava是Reactive Extensions在JVM平台上的一个实现，通过使用观察者序列来构建异步、基于事件的程序。RxJava可以说是观察者设计模式的一个扩展，支持不同的数据/事件流和额外的操作类，允许通过声明式的方式构建不同的执行序列，通过抽象的方式屏蔽底层的多线程实现、同步、线程安全、并发数据结构、非阻塞 I/O 等逻辑。RxJava 支持 Java 5之后的版本，还支持跑在JVM上的各种其他语言，例如Groovy、Clojure、JRuby、Kotlin和 Scala等。笔者建议尽量使用Java 8及以上的版本，因为能够使用Lambda表达式等新特性。
RxJava使用Observables来访问多对象的异步序列。Observables是可组合的，Java Futures模型在用于单一层面的异步执行方面比较方便，但是在需要异步嵌套执行时使用起来却很复杂，至少在Java 8出现之前是这样的。用Futures在构建有条件的异步执行流时会非常困难，因为在运行时每次请求的延迟都是不可控的。虽然理论上能够做到，但是过程会很复杂，有时候会在Future.get（）上一直阻塞，这样的结果显然无法体现异步执行的优势。在 Java 8 之后新增了CompletableFuture，弥补了原先Futures模型的问题。
当然，RxJava不并限于是在Java 8之后，还是之前使用。Observables本身是专门用于构建异步数据流和执行片段的，它很灵活，RxJava的 Observables除了支持Futures模型的所有功能外，还支持无限的数据流；Observables本身是一个统一抽象，用于支持不用情况下的异步数据序列的执行过程。1.3 为何选择RxJava
Rx扩展了观察者模式用于支持数据和事件序列，添加了一些操作符，让你可以使用声明式的方式来组合这些序列，而无须关注底层的实现，如线程、同步、线程安全、并发数据结构和非阻塞I/O。
Rx的Observable模型让开发者可以像使用集合数据一样操作异步事件流，对异步事件流使用各种简单、可组合的操作。
◎ 可组合
对于单层的异步操作来说，Java中Future对象的处理方式非常简单有效，但是一旦涉及嵌套，它们就开始变得异常烦琐和复杂。在Java 8之前，使用Future很难很好地组合带条件的异步执行流程。从另一方面来说，RxJava的被观察者们（Observable/Flowable/Single/Completable/Maybe）一开始就是为组合异步数据流准备的。
◎ 更灵活
RxJava的Observable不仅支持处理单独的标量值（就像Future可以做的），还支持数据序列，甚至是无穷的数据流。Observable是一个抽象概念，适用于任何场景。Observable拥有它的近亲Iterable的全部优雅与灵活。
Observable是异步的双向push，Iterable是同步的单向pull，可以通过表1-1进行对比。表1-1
◎ 无偏见
Rx对于并发性或异步性没有任何特殊的偏好，Observable可以用任何方式（如线程池、事件循环、非阻塞I/O或Actor模式）来满足我们的需求。无论我们选择怎样实现它，无论底层实现是阻塞的还是非阻塞的，客户端代码都将与Observable的全部交互当成是异步的。
1.Observable是如何实现的
public Observable＜T＞getData();
◎ 它能与调用者在同一线程同步执行吗？
◎ 它能异步地在单独的线程执行吗？
◎ 它会将工作分发到多个线程，返回数据的顺序是任意的吗？
◎ 它使用Actor模式而不是线程池吗？
◎ 它使用NIO和事件循环执行异步网络访问吗？
◎ 它使用事件循环将工作线程从回调线程分离出来吗？
从Observer的视角来看，这些都无所谓，重要的是：使用Rx，可以改变我们的观念，可以在完全不影响Observable程序库使用者的情况下，彻底地改变Observable的底层实现。
2.使用回调存在很多问题
回调在不阻塞任何事情的情况下，解决了Future.get（）过早阻塞的问题。响应结果一旦就绪，Callback就会被调用，它们天生就是高效率的。不过，就像使用Future一样，对于单层的异步执行来说，回调很容易使用，对于嵌套的异步组合而言，它们显得非常笨拙。
3.Rx是一个多语言的实现
Rx在大量的编程语言中都有实现，并尊重实现语言的风格，而且更多的实现正在飞速增加。
4.响应式编程
Rx提供了一系列的操作符，我们可以使用它们来过滤（filter）、选择（select）、变换（transform）、结合（combine）和组合（compose）多个Observable，这些操作符让执行和复合变得非常高效。
我们可以把Observable当作Iterable推送方式的等价物，使用Iterable，消费者从生产者那拉取数据，线程阻塞直至数据准备好。使用 Observable，在数据准备好时，生产者将数据推送给消费者。数据可以同步或异步到达，这种方式更加灵活。
下面的例子展示了相似的高阶函数在Iterable和Observable上的应用。
Observable类型给GOF的观察者模式添加了两种缺少的语义，这样就和Iterable类型中可用的操作一致了。
生产者可以发信号给消费者，通知它没有更多数据可用了（对于Iterable，一个for循环正常完成表示没有数据了；对于 Observable，就是调用观察者的 onComplete 方法） 生产者可以发信号给消费者，通知它遇到了一个错误（对于 Iterable，在迭代过程中发生错误会抛出异常；对于Observable，就是调用观察者（Observer）的onError方法）。有了这两种功能，Rx就能使Observable与Iterable保持一致了，唯一的不同是数据流的方向。任何对Iterable的操作，都可以对Observable适用。
再举一个例子，某App从服务端获取酒店列表，并将价格大于等于500元的房间全部列出来展示在界面上。
如果使用RxJava的话，大致可以这样写。
当然，使用Lanbda表达式之后还可以更加简化一点。
其实，下面这两句还能用compose操作符和Transformer一起封装。
在后面的章节中会详细描述，在这里如果没有理解也没有关系。1.4 RxJava能做什么
曾经有人问我，在Android开发中哪些地方可以使用RxJava？我当时给的回答是，任何地方，包括App所依赖的底层框架。
例如，可以使用RxBinding，它是对Android View事件的扩展，可以对View事件使用RxJava的各种操作。
也可以使用RxJava替换原先的EventBus，在Square Otto的GitHub主页上有这样两句话。
"This project is deprecated in favor of RxJava and RxAndroid.These projects permit the same event-driven programming model as Otto,but they're more capable and offer better control of threading."
对于Android的AsyncTask，也可以完全使用RxJava来替代。
这些内容，本书后续的章节都会有详细的介绍。
当然，不只是Android App的开发，在服务端领域的开发中也可以使用RxJava。本书后面几章会有一些例子涉及如何在服务端开发中使用RxJava。
总之，RxJava能做的事情很多，如果你的项目中还没有引入RxJava，那么现在是引入RxJava 2.x的最好时机，它会给你带来新的编程体验，让你感受响应式编程的乐趣。如果你的项目中已经使用了RxJava 1.x，那么强烈建议你升级到最新的版本。RxJava 2 还是使用原来的思想，与RxJava 1 区别不大，从 RxJava 1 迁移到 RxJava 2 成本并不高，也花费不了多少工夫。1.5 RxJava 2的Hello World
学习一门新的语言或者一个新的框架，一般第一次接触时都会写一个Hello World。
下面来看看RxJava版本的Hello World吧。
为什么这个Hello World版本这么长？能不能写得简单一点呢？当然可以啦！
是不是比刚才简单了一些，还想再简单一点吗？那好，我们借助Lambda表达式再来看看。
这次一句话就搞定了，符合Hello World的本意。System.out：：println是方法引用，简化版的Lambda表达式，如果不熟悉Lambda表达式也不用纠结，它是Java 8的特性，在本书的第16章中会有详细介绍，对此感兴趣的读者也可以提前去了解一下。1.6 小结
本章简单介绍了RxJava的历史，RxJava是结合了多种设计模式并优化而产生的结晶。
RxJava 这么好，到底有哪些实际应用呢？目前在移动端可以看到大量使用 RxJava 来编写的代码，也因此产生了大量的优秀框架，比如RxBinding、Retrofit、RxLifecycle等，服务端的开发也可以大量使用RxJava。CouchBase这个NoSQL数据库的Java SDK就是采用RxJava来编写的。除此之外，Netflix的Hystrix的底层也是使用RxJava来编写的，在微服务架构中Hystrix用于帮助隔离每个服务，即使单个服务的响应失败，也不会影响整个请求的响应。
RxJava 给我们带来了一种新的编程思想，异步编程由此变得更加简单。笔者整理出的RxJava 2的思维导图如图1-1所示，本书后面的章节会对这些内容分别讲述。图1-1 第2章 RxJava基础知识2.1 Observable
RxJava的使用通常需要三步。（1）创建Observable
Observable的字面意思是被观察者，使用RxJava时需要创建一个被观察者，它会决定什么时候触发事件以及触发怎样的事件。有点类似上游发送命令，可以在这里决定异步操作模块的顺序和异步操作模块的次数。（2）创建Observer
Observer即观察者，它可以在不同的线程中执行任务。这种模式可以极大地简化并发操作，因为它创建了一个处于待命状态的观察者哨兵，可以在未来某个时刻响应Observable的通知，而不需要阻塞等待Observable发射数据。（3）使用subscribe（）进行订阅
创建了Observable和Observer之后，我们还需要使用subscribe（）方法将它们连接起来，这样整个上下游就能衔接起来实现链式调用。还记得第1章那个RxJava的Hello World吗？
just（）是RxJava的创建操作符，用于创建一个Observable。Consumer 是消费者，用于接收单个值。熟悉Java 8的读者可以看到它与java.util.function.Consumer类似，RxJava 2的命名规范参照Java 8。
subscribe有多个重载的方法。第一个就是刚才的“Hello World”版本。
接下来，我们来看一个重载方法的版本，subscribe（onNext，onError，onComplete）。
执行结果。
此时，除了打印了“Hello World”，还打印了“onComplete（）”。在subscribe中onComplete是执行完onNext之后再执行的。onComplete是一个Action，它与Consumer的区别如下。
◎ Action：无参数类型。
◎ Consumer：单一参数类型。
再来看一个重载方法的版本，subscribe（onNext，onError，onComplete，onSubscribe）。
执行结果：
它先打印了“subscribe”，接着打印了“Hello World”和“onComplete（）”。这次先执行了onSubscribe，再执行了onNext和onComplete。千万不要着急，一会儿先介绍do操作符，do操作符涉及RxJava内部的数据流向。
在RxJava 2 中，Observable不再支持订阅Subscriber，而是需要使用Observer作为观察者，下面代码使用的Observer就等价于刚才的代码。
执行结果：
在RxJava中，被观察者、观察者、subscribe（）方法三者缺一不可。只有使用了subscribe（），被观察者才会开始发送数据，这一点极为重要。
RxJava 2的5种观察者模式如图2-1所示。图2-1
本章的前两个小节主要讲述 Observable 的相关内容。2.3 节以及本书的第 8 章主要讲述Flowable。其余的三种被观察者模式会在2.4节中讲述。对5种观察者模式的描述如表2-1所示。表2-1
从表2-1中可以看出，5种被观察者类型中只有Flowable支持背压，如果有需要背压的情况，则必须使用Flowable。
do操作符
do操作符可以给Observable的生命周期的各个阶段加上一系列的回调监听，当Observable执行到这个阶段时，这些回调就会被触发。在RxJava中包含了很多的doXXX操作符。
例如下面的代码，基本包含了Observable的完整生命周期。
执行结果：
执行结果显示了RxJava的内部数据流向。最开始是doOnSubscribe，等到观察者消费完之后，会执行doFinally、doAfterTerminate。表2-2总结了一些常用的do操作符的用途。表2-2
试读结束[说明：试读内容隐藏了图片]