标签: Observable

RxJava系列 基本概念及使用介绍

前言

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

观察者模式

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

观察者模式很适合下面这些场景中的任何一个:

  1. 当你的架构有两个实体类,一个依赖另一个,你想让它们互不影响或者是独立复用它们时。
  2. 当一个变化的对象通知那些与它自身变化相关联的未知数量的对象时。
  3. 当一个变化的对象通知那些无需推断具体类型的对象时。

通常一个观察者模式的类图是这样的:

%title插图%num
Observer

如果你对观察者模式不是很了解,那么强烈建议你先去学习下。关于观察者模式的详细介绍可以参考我之前的文章:设计模式之观察者模式

扩展的观察者模式

在RxJava中主要有4个角色:

  • Observable
  • Subject
  • Observer
  • Subscriber

Observable和Subject是两个“生产”实体,Observer和Subscriber是两个“消费”实体。说直白点Observable对应于观察者模式中的被观察者,而ObserverSubscriber对应于观察者模式中的观察者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创建了!

大兄弟你等等…,你之前那篇观察者模式中不是说观察者只提供一个update方法的吗?这特么怎么有三个?!!

少年勿急,且听我慢慢道来。在普通的观察者模式中观察者一般只会提供一个update()方法用于被观察者的状态发生变化时,用于提供给被观察者调用。而在RxJava中的观察者Observer提供了:onNext()、 onCompleted()onError()三个方法。还记得吗?开篇我们讲过RxJava是基于一种扩展的观察这模式实现,这里多出的onCompleted和onError正是对观察者模式的扩展。ps:onNext就相当于普通观察者模式中的update

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订阅观察者Observerps:你没看错,不同于普通的观察者模式,这里是被观察者订阅观察者

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

observable.subscribe(observer);
%title插图%num
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<Integer>,并为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();
    }
}

我们重点看看hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber),前面这个hook.onSubscribeStart(observable, observable.onSubscribe)返回的是它自己括号内的第二个参数observable.onSubscribe,然后调用了它的call方法。而这个observable.onSubscribe正是create()方法中的Subscriber,这样整个流程就理顺了。看到这里是不是对RxJava的执行流程清晰了一点呢?这里也建议大家在学习新技术的时候多去翻一翻源码,知其然还要能知其所以然不是吗。

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

异步

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

%title插图%num

同时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(Schedulers.io())指定了获取小区列表、处理房源信息等一系列事件都是在IO线程中运行,observeOn(AndroidSchedulers.mainThread())指定了在屏幕上展示房源的操作在UI线程执行。这就做到了在子线程获取房源,主线程展示房源。

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

RxJava系列(组合操作符)

这一章我们接着介绍组合操作符,这类operators可以同时处理多个Observable来创建我们所需要的Observable。组合操作符主要包含: Merge StartWith Concat Zip CombineLatest SwitchOnNext Join等等。

Merge

merge(Observable, Observable)将两个Observable发射的事件序列组合并成一个事件序列,就像是一个Observable发射的一样。你可以简单的将它理解为两个Obsrvable合并成了一个Observable,合并后的数据是无序的。

%title插图%num
merge(Observable, Observable)

我们看下面的例子,一共有两个Observable:一个用来发送字母,另一个用来发送数字;现在我们需要两连个Observable发射的数据合并。

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);

Observable.merge(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }

            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString()+" ");
            }
        });

程序输出:

A 0 B C 1 D E 2 F 3 G H 4 

merge(Observable[])将多个Observable发射的事件序列组合并成一个事件序列,就像是一个Observable发射的一样。

%title插图%num
merge(Observable[])

StartWith

startWith(T)用于在源Observable发射的数据前插入数据。使用startWith(Iterable<T>)我们还可以在源Observable发射的数据前插入Iterable。官方示意图:

%title插图%num
startWith(T) startWith(Iterable<T>)

startWith(Observable<T>)用于在源Observable发射的数据前插入另一个Observable发射的数据(这些数据会被插入到
源Observable发射数据的前面)。官方示意图:

%title插图%num
startWith(Observable<T>)

Concat

concat(Observable<? extends T>, Observable<? extends T>) concat(Observable<? extends Observable<T>>)用于将多个obserbavle发射的的数据进行合并发射,concat严格按照顺序发射数据,前一个Observable没发射玩是不会发射后一个Observable的数据的。它和merge、startWitch和相似,不同之处在于:

  1. merge:合并后发射的数据是无序的;
  2. startWitch:只能在源Observable发射的数据前插入数据。
    %title插图%num
    concat(Observable<? extends T>, Observable<? extends T>)、concat(Observable<? extends Observable<T>>)

这里我们将前面Merge操作符的例子拿过来,并将操作符换成Concat,然后我们看看执行结果:

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);

Observable.concat(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }

            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString() + " ");
            }
        });

程序输出:

A B C D E F G H 0 1 2 3 4 

Zip

zip(Observable, Observable, Func2)用来合并两个Observable发射的数据项,根据Func2函数生成一个新的值并发射出去。当其中一个Observable发送数据结束或者出现异常后,另一个Observable也将停在发射数据。

%title插图%num
zip(Observable, Observable, Func2)

和前面的例子一样,我们将操作符换成了zip:

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(120, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(200, TimeUnit.MILLISECONDS).take(5);

Observable.zip(letterSequence, numberSequence, new Func2<String, Long, String>() {
    @Override
    public String call(String letter, Long number) {
        return letter + number;
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("Error:" + e.getMessage());
    }

    @Override
    public void onNext(String result) {
        System.out.print(result + " ");
    }
});

程序输出:

A0 B1 C2 D3 E4

CombineLatest

comnineLatest(Observable, Observable, Func2)用于将两个Observale*近发射的数据已经Func2函数的规则进展组合。下面是官方提供的原理图:

%title插图%num
comnineLatest(Observable, Observable, Func2)

下面这张图应该更容易理解:

%title插图%num
comnineLatest(Observable, Observable, Func2)
List<String> communityNames = DataSimulator.getCommunityNames();
List<Location> locations = DataSimulator.getLocations();

Observable<String> communityNameSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return communityNames.get(position.intValue());
            }
        }).take(communityNames.size());
Observable<Location> locationSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, Location>() {
            @Override
            public Location call(Long position) {
                return locations.get(position.intValue());
            }
        }).take(locations.size());

Observable.combineLatest(
        communityNameSequence,
        locationSequence,
        new Func2<String, Location, String>() {
            @Override
            public String call(String communityName, Location location) {
                return "小区名:" + communityName + ", 经纬度:" + location.toString();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }
        });

程序输出:

小区名:竹园新村, 经纬度:(21.827, 23.323)
小区名:康桥半岛, 经纬度:(21.827, 23.323)
小区名:康桥半岛, 经纬度:(11.923, 16.309)
小区名:中粮·海景壹号, 经纬度:(11.923, 16.309)
小区名:中粮·海景壹号, 经纬度:(22.273, 53.623)
小区名:浦江名苑, 经纬度:(22.273, 53.623)
小区名:南辉小区, 经纬度:(22.273, 53.623)

SwitchOnNext

switchOnNext(Observable<? extends Observable<? extends T>>用来将一个发射多个小Observable的源Observable转化为一个Observable,然后发射这多个小Observable所发射的数据。如果一个小的Observable正在发射数据的时候,源Observable又发射出一个新的小Observable,则前一个Observable发射的数据会被抛弃,直接发射新
的小Observable所发射的数据。

结合下面的原理图大家应该很容易理解,我们可以看到下图中的黄色圆圈就被丢弃了。

%title插图%num
switchOnNext(Observable<? extends Observable<? extends T>>)

Join

join(Observable, Func1, Func1, Func2)我们先介绍下join操作符的4个参数:

  • Observable:源Observable需要组合的Observable,这里我们姑且称之为目标Observable;
  • Func1:接收从源Observable发射来的数据,并返回一个Observable,这个Observable的声明周期决定了源Obsrvable发射出来的数据的有效期;
  • Func1:接收目标Observable发射来的数据,并返回一个Observable,这个Observable的声明周期决定了目标Obsrvable发射出来的数据的有效期;
  • Func2:接收从源Observable和目标Observable发射出来的数据,并将这两个数据组合后返回。

所以Join操作符的语法结构大致是这样的:onservableA.join(observableB, 控制observableA发射数据有效期的函数, 控制observableB发射数据有效期的函数,两个observable发射数据的合并规则)

join操作符的效果类似于排列组合,把*个数据源A作为基座窗口,他根据自己的节奏不断发射数据元素,第二个数据源B,每发射一个数据,我们都把它和*个数据源A中已经发射的数据进行一对一匹配;举例来说,如果某一时刻B发射了一个数据“B”,此时A已经发射了0,1,2,3共四个数据,那么我们的合并操作就会把“B”依次与0,1,2,3配对,得到四组数据: [0, B] [1, B] [2, B] [3, B]

再看看下面的图是不是好理解了呢?!

%title插图%num
join(Observable, Func1, Func1, Func2)

读懂了上面的文字,我们再来写段代码加深理解。

final List<House> houses = DataSimulator.getHouses();//模拟的房源数据,用于测试

//用来每秒从houses总取出一套房源并发射出去
Observable<House> houseSequence =
        Observable.interval(1, TimeUnit.SECONDS)
                .map(new Func1<Long, House>() {
                    @Override
                    public House call(Long position) {
                        return houses.get(position.intValue());
                    }
                }).take(houses.size());//这里的take是为了防止houses.get(position.intValue())数组越界

//用来实现每秒发送一个新的Long型数据
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);

houseSequence.join(tictoc,
        new Func1<House, Observable<Long>>() {
            @Override
            public Observable<Long> call(House house) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        },
        new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                return Observable.timer(0, TimeUnit.SECONDS);
            }
        },
        new Func2<House, Long, String>() {
            @Override
            public String call(House house, Long aLong) {
                return aLong + "-->" + house.getDesc();
            }
        }
).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("Error:"+e.getMessage());
    }

    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
});

程序输出:

0-->中粮海景壹号新出大平层!总价4500W起
1-->中粮海景壹号新出大平层!总价4500W起
1-->满五唯一,黄金地段
2-->中粮海景壹号新出大平层!总价4500W起
2-->满五唯一,黄金地段
2-->一楼自带小花园
3-->一楼自带小花园
3-->毗邻汤臣一品
4-->毗邻汤臣一品
4-->顶级住宅,给您总统般尊贵体验
5-->顶级住宅,给您总统般尊贵体验
5-->顶层户型,两室一厅
6-->顶层户型,两室一厅
6-->南北通透,豪华五房
7-->南北通透,豪华五房

通过转换操作符、过滤操作符、组合操作符三个篇幅将RxJava主要的操作符也介绍的七七八八了。更多操作符的介绍建议大家去查阅官方文档,并自己动手实践一下。这一系列的文章也会持续更新,欢迎大家保持关注!:)

友情链接: SITEMAP | 旋风加速器官网 | 旋风软件中心 | textarea | 黑洞加速器 | jiaohess | 老王加速器 | 烧饼哥加速器 | 小蓝鸟 | tiktok加速器 | 旋风加速度器 | 旋风加速 | quickq加速器 | 飞驰加速器 | 飞鸟加速器 | 狗急加速器 | hammer加速器 | trafficace | 原子加速器 | 葫芦加速器 | 麦旋风 | 油管加速器 | anycastly | INS加速器 | INS加速器免费版 | 免费vqn加速外网 | 旋风加速器 | 快橙加速器 | 啊哈加速器 | 迷雾通 | 优途加速器 | 海外播 | 坚果加速器 | 海外vqn加速 | 蘑菇加速器 | 毛豆加速器 | 接码平台 | 接码S | 西柚加速器 | 快柠檬加速器 | 黑洞加速 | falemon | 快橙加速器 | anycast加速器 | ibaidu | moneytreeblog | 坚果加速器 | 派币加速器 | 飞鸟加速器 | 毛豆APP | PIKPAK | 安卓vqn免费 | 一元机场加速器 | 一元机场 | 老王加速器 | 黑洞加速器 | 白石山 | 小牛加速器 | 黑洞加速 | 迷雾通官网 | 迷雾通 | 迷雾通加速器 | 十大免费加速神器 | 猎豹加速器 | 蚂蚁加速器 | 坚果加速器 | 黑洞加速 | 银河加速器 | 猎豹加速器 | 海鸥加速器 | 芒果加速器 | 小牛加速器 | 极光加速器 | 黑洞加速 | movabletype中文网 | 猎豹加速器官网 | 烧饼哥加速器官网 | 旋风加速器度器 | 哔咔漫画 | PicACG | 雷霆加速