RxJava(一)、初识RxJava

澹台昆

2023-12-01

RxJava

"是什么，为什么，怎么办”是认识问题的逻辑思维过程,这篇文章也是这样来解释RxJava的

RxJava是什么？

RxJava 在 GitHub 主页上的自我介绍是 “a library for composing asynchronous and event-based programs using observable sequences for the Java VM”（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。这就是 RxJava ，概括得非常精准。

其实， RxJava 的本质可以压缩为异步这一个词。说到根上，它就是一个实现异步操作的库，而别的定语都是基于这之上的。

为什么要使用RxJava？

RxJava 的本质可以压缩为异步这一个词。它就是一个实现异步操作的库。

但是Android提供了AsyncTask和Handler也可以实现异步，它们有什么区别呢？

AsyncTask相比RxJava有以下缺点:

书写复杂
异常处理困难（try/catch?）
Activity/Fragment的生命周期导致AsyncTask有不可预见的问题
无法足够简单的做复杂的异步（串行异步/并行异步同步UI）
难以测试
异步数据无法得到良好的缓存

RxJava则对这些问题提供了解决方法：

关于书写复杂的问题:

RxJava结合lambda是一个很好的解决方案

webService.doSomething(someData)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        result -> resultText.setText("It worked!")),
        e -> handleError(e)
    );

关于异常处理困难的问题:

RxJava 中所有的错误都会回调到onError，使用了观察者模式

Subscriber<String> mySubscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) { System.out.println(s); }

// 如果正确的终结，最后会调到这里
    @Override
    public void onCompleted() { }

// 只要有异常抛出（包括操作符中的调用），会调到这里
    @Override
    public void onError(Throwable e) { }
};

关于Activity/Fragment生命周期导致AsyncTask难以维护的问题:

RxAndroid也给出了很好的解决方案

AppObservable.bindFragment(this, webService.doSomething(someData))
   .subscribe(
       result -> resultText.setText("It worked!")),
       e -> handleError(e)
   );

关于AsyncTask无法足够简单做复杂的异步的问题:

RxJava中的各类”操作”可以解决这个问题，这里如果有较多的线程切换，可以考虑使用compose

//这里是一个链式Web Service调用的例子，这些请求互相依赖，在线程池中运行第二批并行调用，然后在将结果返回给Observer之前，对数据进行合并和排序。
public Observable<List<CityWeather>> getWeatherForLargeUsCapitals() {
    return cityDirectory.getUsCapitals()
        .flatMap(cityList -> Observable.from(cityList))
        .filter(city -> city.getPopulation() > 500,000)
        .flatMap(city -> weatherService.getCurrentWeather(city)) //each runs in parallel
        .toSortedList((cw1,cw2) -> cw1.getCityName().compare(cw2.getCityName()));
    }

关于AsyncTask难以测试:

RxJava通过toblocking()将一个异步方法变为同步方法来完成测试

List results = getWeatherForLargeUsCapitals().toBlocking().first();
assertEquals(12, results.size());

关于AsyncTask异步数据无法得到良好的缓存

通过RxAndroid提供的方法，保存一个对Observable 的缓存的引用

@Override
   public void onCreate(Bundle savedInstanceState) {
       super.onCreate(savedInstanceState);

       setRetainInstance(true);
       weatherObservable = weatherManager.getWeatherForLargeUsCapitals().cache();
   }

   public void onViewCreated(...) {
       super.onViewCreated(...)
       bind(weatherObservable).subscribe(this);
   }

.它更大的优点在于：随着程序逻辑变得越来越复杂，它依然能够保持简洁。

RxJava 的实现，是一条从上到下的链式调用，没有任何嵌套，这在逻辑的简洁性上是具有优势的。当需求变得复杂时，这种优势将更加明显。

RxJava的使用

RxJava 的异步实现，是通过一种扩展的观察者模式来实现的。那什么是观察者模式呢？

一、观察者模式
RxJava 有四个基本概念：
Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。
Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。

与传统观察者模式不同， RxJava 的事件回调方法除了普通事件 onNext() （相当于 onClick() / onEvent()）之外，还定义了两个特殊的事件：onCompleted() 和 onError()。

onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

二、RxJava 的基本实现主要有三点：

创建观察者

Observer 即观察者，它决定事件触发的时候将有怎样的行为。 RxJava 中的 Observer 接口的实现方式：

Observer<String> observer = new Observer<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

除了 Observer 接口之外，RxJava 还内置了一个实现了 Observer 的抽象类：Subscriber。 Subscriber 对 Observer 接口进行了一些扩展，但他们的基本使用方式是完全一样的：

RxJava(一)、初识RxJava

RxJava

RxJava是什么？

为什么要使用RxJava？

RxJava的使用

相关阅读

相关文章

相关问答

相关文档