前言

上一篇我們了解了RxJava的介紹以及通過一個(gè)例子學(xué)習(xí)了RxJava的展現(xiàn)形式奶赔，相信大家現(xiàn)在已經(jīng)對(duì)RxJava有一個(gè)初步的印象與概念了栓票。本篇我們將更深入的學(xué)習(xí)RxJava的基本原理以及如何使用RxJava。

觀察者模式

上一篇有提到RxJava的核心便是擴(kuò)展的觀察者模式，我們先來學(xué)習(xí)一下觀察者模式。

使用

了解了觀察者模式蛇更，接下來動(dòng)起指尖嘗試下怎么使用RxJava。

創(chuàng)建被觀察者Observable

第一步：創(chuàng)建被觀察者Observable：

Observable<Object> observable = Observable.create(new Observable.OnSubscribe<Object>() {
    @Override
    public void call(Subscriber<? super Object> subscriber) {
        //被觀察者的數(shù)據(jù)操作更新
        for (int i = 0; i < 5; i++)
            subscriber.onNext("xulei" + i);
        }
        subscriber.onCompleted();
    }
});

通過Observable.create()方法創(chuàng)建并返回一個(gè)Observable對(duì)象赛糟，create()方法需要一個(gè)OnSubsribe對(duì)象作為參數(shù)傳入并執(zhí)行call()方法派任。此OnSubsribe對(duì)象繼承自Action1。

而在call()方法中可通過subscriber.onNext()進(jìn)行事件的發(fā)送璧南，subscriber.onCompleted()標(biāo)記為事件發(fā)送完畢掌逛。subscriber.onError()標(biāo)記為事件發(fā)送過程出現(xiàn)異常。

除create()方法創(chuàng)建Observable方式外司倚，還可以使用just()與from()等方式豆混，這個(gè)在之后的創(chuàng)建操作符文章中會(huì)有講解。

創(chuàng)建觀察者Observer

第二步：創(chuàng)建觀察者Observer：

Observer observer = new Observer() {
    @Override
    public void onCompleted() { //標(biāo)記事件發(fā)送完畢的回調(diào)
        Log.e("rx_test", "onCompleted");
    }
    @Override
    public void onError(Throwable e) { //事件發(fā)送過程出現(xiàn)異常的回調(diào)
        Log.e("rx_test", "onError");
    }
    @Override
    public void onNext(Object o) { //接收到發(fā)送事件的回調(diào)
        Log.e("rx_test", "onNext:" + o.toString());
    }
};

在以往普通的觀察者模式中动知，觀察者通常只提供一個(gè)update()的方法皿伺，而RxJava擴(kuò)展的觀察者模式中的觀察者提供onNext(Object o)，onCompleted()盒粮，onError(Throwable e)三個(gè)方法鸵鸥，分別對(duì)應(yīng)被觀察者call()方法中重寫的subscriber.onNext()，subscriber.onCompleted()，subscriber.onError()方法妒穴。而onNext()方法就對(duì)應(yīng)普通觀察者模式中的update()方法宋税。

RxJava擴(kuò)展的觀察模式彌補(bǔ)了普通觀察者模式的不足：

1. 當(dāng)事件發(fā)送完畢可以調(diào)用onCompleted()方法進(jìn)行標(biāo)示。
2. 當(dāng)事件發(fā)送過程中出現(xiàn)異常會(huì)自動(dòng)觸發(fā)onError()方法讼油，也可進(jìn)行手動(dòng)調(diào)用杰赛。
3. Observable支持鏈?zhǔn)骄幊蹋瑥亩苊饬嘶卣{(diào)嵌套的問題矮台，也簡(jiǎn)化了代碼乏屯。

訂閱

最后一步：被觀察者Observable訂閱觀察者Observer。

這里不同于普通觀察者模式嘿架，上一篇我們講了在RxJava擴(kuò)展觀察者模式中瓶珊，如果沒有觀察者，被觀察者是不會(huì)發(fā)出任何事件的耸彪。所以這里是被觀察者訂閱觀察者伞芹。

observable.subscribe(observer);

訂閱很簡(jiǎn)單，一行代碼搞定蝉娜。調(diào)用Observable的subscribe()方法唱较，傳入觀察者對(duì)象作為參數(shù)，這樣被觀察者與觀察者就進(jìn)行了綁定召川。

簡(jiǎn)寫

RxJava可以使用鏈?zhǔn)骄幊蹋?/p>

Observable.create(new Observable.OnSubscribe<Object>() {
    @Override
    public void call(Subscriber<? super Object> subscriber) {
        //被觀察者的數(shù)據(jù)操作更新
        for (int i = 0; i < 5; i++) {
            subscriber.onNext("xulei" + i);
        }
        subscriber.onCompleted();
    }
}).subscribe(new Observer<Object>() {
    @Override
    public void onCompleted() {
        Log.e("rx_test", "onCompleted");
    }
    @Override
    public void onError(Throwable e) {
        Log.e("rx_test", "onError");
    }
    @Override
    public void onNext(Object o) {
        Log.e("rx_test", "onNext:" + o.toString());
    }
});

邏輯梳理

到此一個(gè)RxJava的簡(jiǎn)單調(diào)用就實(shí)現(xiàn)了南缓，相信不少碼友看完之后還是很懵13，那我們?cè)賮硎崂硪幌抡麄€(gè)流程荧呐。首先使用Observable.create()創(chuàng)建了一個(gè)Observable<Integer>汉形，并在create()方法中傳入了一個(gè)OnSubscribe對(duì)象作為參數(shù)，重寫OnSubscribe中的call()方法倍阐，當(dāng)Observable進(jìn)行訂閱后就會(huì)自動(dòng)觸發(fā)call()方法發(fā)送事件概疆。

注意：call()方法中的subscriber參數(shù)其實(shí)就是我們?cè)谡{(diào)用subscribe()進(jìn)行訂閱時(shí)時(shí)傳入的觀察者Observer。所以當(dāng)在call()方法中調(diào)用的5次onNext()方法與1次onCompleted()方法后峰搪，觀察者Observer中輸出的日志為：

onNext:xulei0
onNext:xulei1
onNext:xulei2
onNext:xulei3
onNext:xulei4
onCompleted

有的碼友看到這里會(huì)對(duì)call()方法中的subscriber參數(shù)就是訂閱時(shí)傳入的Observer有些不解岔冀，那么我們一起來看源碼。

先看下call()方法中的參數(shù)Subscriber：

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

可以看出Subscriber是Observer的抽象實(shí)現(xiàn)類概耻，與Observer一個(gè)類型使套。再來看訂閱時(shí)調(diào)用的subscribe(observer)方法：

public final Subscription subscribe(final Observer<? super T> observer) {
    //無視
    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);
        }
    });
}

這里將傳入的Observer對(duì)象通過代理轉(zhuǎn)換為Subscriber對(duì)象作為參數(shù)傳入并調(diào)用重載方法subscribe()，我們?cè)賮砜催@個(gè)subscribe()方法的代碼：

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

這里繼續(xù)調(diào)用重載方法subscribe()鞠柄，將由Observer轉(zhuǎn)換為的subscriber對(duì)象與當(dāng)前Observable對(duì)象作為參數(shù)傳入侦高，繼續(xù)往下追蹤源碼：

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();
    }
}

精簡(jiǎn)一下代碼發(fā)現(xiàn)，subscriber.onStart()用來通知被觀察者準(zhǔn)備開始發(fā)送事件了厌杜。重點(diǎn)看hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);我們發(fā)現(xiàn)hook.onSubscribeStart(observable, observable.onSubscribe)方法返回的是其第二個(gè)參數(shù)observable.onSubscribe矫膨，這個(gè)參數(shù)正是新建Observable時(shí)create()方法中傳入的OnSubscribe對(duì)象。接著又通過這個(gè)onSubscribe調(diào)用call(subscriber)方法，將subscriber作為參數(shù)傳入侧馅，碼友們有沒有發(fā)現(xiàn)危尿，這個(gè)subscriber傳來傳去究其根源，正是一開始由訂閱方法subscribe(observer)傳入的observer通過代理轉(zhuǎn)換為的那個(gè)subscriber馁痴。

這樣前后邏輯就連接上了谊娇，通過閱讀源碼我們才能發(fā)現(xiàn)為何創(chuàng)建Observable時(shí)call()方法中的subscriber參數(shù)其實(shí)就是訂閱時(shí)傳入的Observer。

總結(jié)

到此罗晕，本篇關(guān)于RxJava的基本原理與使用就講解完畢了济欢，下一篇我們將一起研究RxJava的四類操作符中的創(chuàng)建操作符都有哪些以及如何使用。

這里感謝下張磊大神撰寫的RxJava系列的幫助：
RxJava系列2(基本概念及使用介紹)

技術(shù)渣一枚小渊，有寫的不對(duì)的地方歡迎大神們留言指正法褥，有什么疑惑或者建議也可以在我Github上RxJavaDemo項(xiàng)目Issues中提出，我會(huì)及時(shí)回復(fù)酬屉。

附上RxJavaDemo的地址：
RxJavaDemo