RxJava使用说明2-基本概念及使用介绍

约定不等于承诺〃 2022-10-30 05:14 243阅读 0赞

*  [RxJava使用说明1-简介][RxJava_1-]
 *  [RxJava使用说明2-基本概念及使用介绍][RxJava_2-]
 *  [RxJava使用说明3-转换操作符][RxJava_3-]
 *  [RxJava使用说明4-过滤操作符][RxJava_4-]
 *  [RxJava使用说明5-组合操作符][RxJava_5-]
 *  [RxJava2更新说明][RxJava2]

# 前言 #

上一篇的示例代码中大家一定发现了Observable这个类。从纯Java的观点看，Observable类源自于经典的观察者模式。RxJava的异步实现正是基于观察者模式来实现的，而且是一种扩展的观察者模式。

# 观察者模式 #

观察者模式基于Subject这个概念，Subject是一种特殊对象，又叫做主题或者被观察者。当它改变时那些由它保存的一系列对象将会得到通知，而这一系列对象被称作Observer(观察者)。它们会对外暴漏了一个通知方法(比方说update之类的)，当Subject状态发生变化时会调用的这个方法。

# 扩展的观察者模式 #

在RxJava中主要有4个角色：

*  Observable
 *  Subject
 *  Observer
 *  Subscriber

Observable和Subject是两个“生产”实体，Observer和Subscriber是两个“消费”实体。说直白点Observable对应于观察者模式中的被观察者，而Observer和Subscriber对应于观察者模式中的观察者。Subscriber其实是一个实现了Observer的抽象类，后面我们分析源码的时候也会介绍到。Subject比较复杂，以后再分析。

上一篇文章中我们说到RxJava中有个关键概念：事件。观察者Observer和被观察者Observable通过subscribe()方法实现订阅关系。从而Observable 可以在需要的时候发出事件来通知Observer。

# RxJava如何使用 #

我自己在学习一种新技术的时候通常喜欢先去了解它是怎么用的，掌握了使用方法后再去深挖其原理。那么我们现在就来说说RxJava到底该怎么用。

## 第一步：创建观察者Observer ##

Observer<Object> observer = new Observer<Object>() { 
    
        @Override
        public void onCompleted() { 
    
        }
    
        @Override
        public void onError(Throwable e) { 
    
        }
    
        @Override
        public void onNext(Object s) { 
    
        }
     };

这么简单，一个观察者Observer创建了!

RxJava中添加了普通观察者模式缺失的三个功能：

1.  RxJava中规定当不再有新的事件发出时，可以调用onCompleted()方法作为标示；当
2.  事件处理出现异常时框架自动触发onError()方法；
3.  同时Observables支持链式调用，从而避免了回调嵌套的问题。

## 第二步：创建被观察者Observable ##

Observable.create()方法可以创建一个Observable，使用crate()创建Observable需要一个OnSubscribe对象，这个对象继承Action1。当观察者订阅我们的Observable时，它作为一个参数传入并执行call()函数。

Observable<Object> observable = Observable.create(new Observable.OnSubscribe<Object>() {          
        @Override
        public void call(Subscriber<? super Object> subscriber) { 
    
        }
    });

除了create()，just()和from()同样可以创建Observable。看看下面两个例子：

just(T…)将传入的参数依次发送

Observable observable = Observable.just("One", "Two", "Three");
    //上面这行代码会依次调用
    //onNext("One");
    //onNext("Two");
    //onNext("Three");
    //onCompleted();

from(T\[\])/from(Iterable<? extends T>)将传入的数组或者Iterable拆分成Java对象依次发送

String[] parameters = { "One", "Two", "Three"};
    Observable observable = Observable.from(parameters);
    //上面这行代码会依次调用
    //onNext("One");
    //onNext("Two");
    //onNext("Three");
    //onCompleted();

## 第三步：被观察者Observable订阅观察者Observer ##

你没看错，不同于普通的观察者模式，这里是被观察者订阅观察者

有了观察者和被观察者，我们就可以通过subscribe() 来实现二者的订阅关系了。

observable.subscribe(observer);

连在一起写就是这样：

Observable.create(new Observable.OnSubscribe<Integer>() { 
    
        @Override
        public void call(Subscriber<? super Integer> subscriber) { 
            for (int i = 0; i < 5; i++) { 
                subscriber.onNext(i);
            }
            subscriber.onCompleted();
        }
    
    }).subscribe(new Observer<Integer>() { 
    
        @Override
        public void onCompleted() { 
            System.out.println("onCompleted");
        }
    
        @Override
        public void onError(Throwable e) { 
            System.out.println("onError");
        }
    
        @Override
        public void onNext(Integer item) { 
            System.out.println("Item is " + item);
        }
    });

至此一个完整的RxJava调用就完成了。

兄台，你叨逼叨叨逼叨的说了一大堆，可是我没搞定你特么到底在干啥啊？！！不急，我现在就来告诉你们到底发生了什么。

首先我们使用Observable.create()创建了一个新的Observable，并为create()方法传入了一个OnSubscribe，OnSubscribe中包含一个call()方法，一旦我们调用subscribe()订阅后就会自动触发call()方法。call()方法中的参数Subscriber其实就是subscribe()方法中的观察者Observer。我们在call()方法中调用了5次onNext()和1次onCompleted()方法。一套流程周下来以后输出结果就是下面这样的：

Item is 0
    Item is 1
    Item is 2
    Item is 3
    Item is 4
    onCompleted

看到这里可能你又要说了，大兄弟你别唬我啊！OnSubscribe的call()方法中的参数Subscriber怎么就变成了subscribe()方法中的观察者Observer？！！！这俩儿货明明看起来就是两个不同的类啊。

我们先看看Subscriber这个类：

public abstract class Subscriber<T> implements Observer<T>, Subscription { 
    
        ...
    }

从源码中我们可以看到，Subscriber是Observer的一个抽象实现类，所以我首先可以肯定的是Subscriber和Observer类型是一致的。接着往下我们看看subscribe()这个方法：

public final Subscription subscribe(final Observer<? super T> observer) { 
    
        //这里的if判断对于我们要分享的问题没有关联，可以先无视
        if (observer instanceof Subscriber) { 
            return subscribe((Subscriber<? super T>)observer);
        }
        return subscribe(new Subscriber<T>() { 
    
            @Override
            public void onCompleted() { 
                observer.onCompleted();
            }
    
            @Override
            public void onError(Throwable e) { 
                observer.onError(e);
            }
    
            @Override
            public void onNext(T t) { 
                observer.onNext(t);
            }
    
        });
    }

我们看到subscribe()方法内部首先将传进来的Observer做了一层代理，将它转换成了Subscriber。我们再看看这个方法内部的subscribe()方法：

public final Subscription subscribe(Subscriber<? super T> subscriber) { 
        return Observable.subscribe(subscriber, this);
    }

进一步往下追踪看看return后面这段代码到底做了什么。精简掉其他无关代码后的subscribe(subscriber, this)方法是这样的：

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) { 
    
        subscriber.onStart();
        try { 
            hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
            return hook.onSubscribeReturn(subscriber);
        } catch (Throwable e) { 
            return Subscriptions.unsubscribed();
        }
    }

> subscribe()的参数除了可以是Observer和Subscriber以外还可以是Action1、Action0；这是一种更简单的回调，只有一个call(T)方法；由于太简单这里就不做详细介绍了！

# 异步 #

上一篇文章中开篇就讲到RxJava就是来处理异步任务的。但是默认情况下我们在哪个线程调用subscribe()就在哪个线程生产事件，在哪个线程生产事件就在哪个线程消费事件。那怎么做到异步呢？RxJava为我们提供Scheduler用来做线程调度，我们来看看RxJava提供了哪些Scheduler。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hndzEwMTA_size_16_color_FFFFFF_t_70]  
同时RxJava还为我们提供了subscribeOn()和observeOn()两个方法来指定Observable和Observer运行的线程。

Observable.from(getCommunitiesFromServer())
                .flatMap(community -> Observable.from(community.houses))
                .filter(house -> house.price>=5000000)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(this::addHouseInformationToScreen);

上面这段代码大家应该有印象吧，没错正是我们上一篇文章中的例子。subscribeOn(http://Schedulers.io())指定了获取小区列表、处理房源信息等一系列事件都是在IO线程中运行，observeOn(AndroidSchedulers.mainThread())指定了在屏幕上展示房源的操作在UI线程执行。这就做到了在子线程获取房源，主线程展示房源。

好了，RxJava系列的入门内容我们就聊到这。下一篇我们再继续介绍更多的API以及它们内部的原理。

[RxJava_1-]: https://blog.csdn.net/xgw1010/article/details/113834578
[RxJava_2-]: https://blog.csdn.net/xgw1010/article/details/113835048
[RxJava_3-]: https://blog.csdn.net/xgw1010/article/details/113836968
[RxJava_4-]: https://blog.csdn.net/xgw1010/article/details/113843665
[RxJava_5-]: https://blog.csdn.net/xgw1010/article/details/113845870
[RxJava2]: https://blog.csdn.net/xgw1010/article/details/113849685
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hndzEwMTA_size_16_color_FFFFFF_t_70]: /images/20221024/f64a019a92e9498eaca2f4930cb1534b.png