RxJava学习以及在机票国内主系统重构经验分享

张楚宸，Java开发工程师，2016年加入去哪儿网。接触过Android、RN、Angular等大前端技术。现阶段主要从事数据算法工程相关工作。

系统中有没有下面的问题存在?

代码复杂，流程不清晰

1、系统逻辑复杂； 2、业务多导致系统代码不断累加； 3、新人上手困难； 4、需要频繁梳理老逻辑。

业务变化快

业务新增、业务间依赖变化，导致需要频繁调整大量代码。

异步嵌套回调

需要使用异步并行的方式缩减请求时长，但是异步嵌套回调的写法让人生厌。

是否需要这样一个框架?

RxJava+Lambda+StreamAPI

简洁的语法，强大的功能

短小精悍又丰富的函数应对大多数应用场景。

可读、可维护、可拓展

1、代码耦合低； 2、可复用性高； 3、数据扭转清晰； 4、新手学习成本降低； 5、业务变动改动量小。

便利的异步调用

轻松异步并获取结果，不再受到回调和阻塞困扰。

什么是RxJava？

ReactiveX

1、ReactiveX 的历史: ReactiveX 是 Reactive Extensions 的缩写,一般简写为 Rx，最初是 LINQ 的一个扩展,由微软的架构师 Erik Meijer 领导的团队开发，在 2012 年 11 月开源，Rx 是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便的处理异步数据流。 2、RX 是为异步事件创建的，也是专门为聚合构建的。

RxJava

1、微软给的定义是：Rx 是一个函数库，让开发者可以利用可观察序列和 LINQ 风格查询操作符来编写异步和基于事件的程序。 2、ReactiveX.io 给的定义是：Rx 是一个使用可观察数据流进行异步编程的编程接口。 3、RxJava 是 ReactiveX 在 Java 上的实现。

RxJava成熟并且社区活跃度高

RxJava2

相比较RxJava1来说，RxJava2： （1）遵循 Reactive-Streams 规范，重新实现。 （2）规避 Null。

RxJava的优势和适用场景

框架特性

1、可以简化代码； 2、可以作为异步框架； 3、可以配合任何框架适用。

函数式风格

对可观察数据流使用无副作用的输入输出函数，避免了程序里错综复杂的状态。

简化代码

对可观察数据流使用无副 Rx 的操作符通通常可以将作用的输入输出函数，避免了复杂的难题简化为很少的几行程序。

异步错误处理

传统的 try/catch 没办法处理异步计算，Rx 提供了合适的错误处理机制。

轻松使用并发

Rx 的 Observables 和 Schedulers 让开发者可以摆脱底层的线程同步和各种并发问题。

响应式编程

1、响应式编程是一种面向数据流和变化传播的编程范式。 2、使用观察者模式。 3、面向数据流编程： （1）创建：可以方便的创建事件流和数据流； （2）组合：使用查询式的操作符组合和变换数据流； （3）监听：可以订阅任何可观察的数据流并执行操作。 4、RxJava 中每一个操作符都对应一个发布-订阅，传递数据的过程。

使用的系统和场景

业务逻辑,变化快

（1）整个系统是高度可组合的； （2）复杂的逻辑可以被分解代码低耦合； （3）可组合,可复用性高。

异步场景较多

（1）有效减少嵌套异步回调； （2）灵活切换线程,执行异步任务； （3）可对异步任务设置超时、异常处理等。

代码复杂度高,可维护性差

（1）函数式风格； （2）响应式编程,提供大量便利的操作符。

聚合型系统

（1）调用外部系统做业务聚合； （2）调用外部接口多。

学习RxJava

学习方法

学习成本略陡峭，但资料较多

编程思想的转换——响应式编程。 资料： （1）API 和教程——供新手入门； （2）拆轮子+源码——了解内部实现。

常用操作符介绍以及使用示例

Just

创建一个发射指定值的 Observable。

Filter

操作符使用你指定的一个谓词函数测试数据项，只有通过测试的数据才会被发射。

Map

基本作用就是将一个 Observable 通过某种函数关系，转换为另一种 Observable。

Zip

操作符返回一个 Obversable，它使用这个函数按顺序结合两个或多个 Observables 发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个 Observable 一样多的数据。

FlatMap

将一个发射数据的 Observable 变换为多个 Observables，然后将它们发射的数据合并后放进一个单独的 Observable。

Timeout

对原始 Observable 的一个镜像，如果过了一个指定的时长仍没有发射数据，它会发一个错误通知。如果原始 Observable 过了指 定的一段时长没有发射任何数据，Timeout 操作符会以一个 onError 通知终止这个 Observable。

Catch

OnErrorReturn 当发生错误的时候，让 Observable 发射一个预先定义好的数据并正常地终止。

线程切换

使用 subscribeOn 和 ObserverOn 来指定执行线程。 通过创建 SubscribeOnObserver 和 ObserveOnSingleObserver，将原有的 Observer 运行在指定的 Scheduler 中。

获取多任务异步结果

使用 zip 聚合多个异步结果。

实现方式

（1）异步任务和同步获取处理后结果的关系：发布-订阅； （2）每个异步任务完成后，计数器-1； （3）计数器为0时，获取到全部任务结果，处理结果，并推送给观察者。

使用RxJava过程的经验分享

学习成本高

问题

一个团队内协作开发时，从零学习 RxJava 成本较高。

解决方案

我们可以通过简单封装，只需写同步代码后使用异步工具类，即可返回异步调用结果。通过这种方式降低团队内部的学习成本。 当然想更好的利用工具必须要对其有一定的理解，在降低学习成本进入开发后还是有必要花时间学习的。

线程控制

问题

RxJava 中默认有5个线程池并封装成了 Scheduler，如果不指定 Scheduler 所有的 Java 都会执行在默认的线程池中。可能有如下问题存在： （1）在 IO 情况下共用线程池有可能存在问题； （2）需要在线程间传递信息时，可能会需要使用自定义封装的线程池； （3）在上面提到的操作符中，很多操作符会创建一个新的数据流，如果没有指定线程池将执行在默认线程池中，造成线程池不可控。

解决方案

解决方案一 （1）可以实现在自己的 Scheduler，人为指定代码运行的线程池； （2）理解操作符的原理，做好封装。 解决方案二 （1）替换默认线程池； （2）RxJava 没有提供入口，需要自己实现替换。

一些建议

1、如果需要跑在自定义线程池中建议仅把 RxJava 作为异步框架使用。 2、如果整个请求链路都使用RxJava的话： （1）需要维护线程池； （2）不停切换线程池带来不必要的线程切换开销。

异常捕获

RxJava 虽然提供了 Catch 操作符用来捕获异常，但是会缺失异步线程中的栈信息。 尽量在同步代码中捕获异常。

关于Debug

因为 RxJava 使用了观察者模式，所以只有当数据流实际产生时才会有消费，真正的逻辑执行是消费时，而不是定义数据流的代码。 需要在实际执行处 Debug，而不是创建 RxJava 数据流的地方 Debug。

项目中收益

在机票国内主系统重构中，我们使用了 RxJava。

代码可读性、可维护性增强

（1）新人上手成本降低； （2）新需求迭代更快速； （3）可以更快定位问题。

流程更清晰

有了友好的异步支持，将流程尽量异步化，提升了性能

当然，使用 RxJava 本身不能提高性能。在异步化的基础上，我们还做了： （1）业务梳理； （2）集群拆分和合并； （3）缓存优化； （4）使用G1作为垃圾收集器； （5）提高单机性能； 新老系统单机压测对比。

总结

使用RxJava2+Java8+Servlet3后可以获得的收益如下：

1、数据走向清晰：面向数据流。 2、代码可读性高，可维护性高：简化代码。 3、异步操作便利：避免回调地狱；切换线程池便利。 4、可拓展性高：每个异步任务独立，可组合复用，可拓展。

使用RxJava作为异步框架

（1）有效并行执行任务，缩短响应时长； （2）阻塞任务异步执行，提高吞吐； （3）异步任务组合灵活，应对较快的业务变化。

使用Lambda+StreamAPI

（1）大量 StreamAPI 应对不同场景需要； （2）有效简洁代码； （3）可读性高。

使用Servlet异步

释放 http 线程资源。