扔物線大大的文章確實寫的牛 扔物線，看了他的文章受益匪淺，文中很多會引用到他的一些分析，沒有看過他的文章的建議先看一下。

一.概述

先簡單介紹一下RxJava的思想

RxJava 有四個基本概念：Observable (可觀察者，即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。Observable 和 Observer 通過 subscribe() 方法實現訂閱關系，從而 Observable 可以在需要的時候發出事件來通知 Observer。

概念介紹

1.Observable（可觀察者，即被觀察者）

事件的觸發者。

2.Observer/Subscriber（觀察者）

事件的產生者

3.subscribe(訂閱)

可被觀察者和觀察者之間的橋梁

4.事件

產生的事件

5.總結

舉個例子：我是一名讀者雜志會員，我想訂閱讀者期刊，當我訂閱之后，讀者工作室就會每個月給我發一本讀者雜志。在這個事件中，我就是一名被觀察者，讀者工作室就是觀察者，因為我一旦產生訂閱這件事，就會觸發讀者工作室的一系列動作。

二.實例解析

1.最簡單的例子

observable(可被觀察者)

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {subscriber.onNext("test1");subscriber.onNext("test2");subscriber.onCompleted();}});

subscriber/observer(觀察者)

private Subscriber<String> subscribe = new Subscriber<String>() {@Overridepublic void onCompleted() {Log.e("HP", "onCompleted");}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(String s) {Log.e("HP", s);} };

subscribe(訂閱)

observable.subscribe(subscribe);

執行以上代碼會打印出如下結果

test1 test2 onCompleted

這樣一個最簡單的RxJava代碼就完成了。

為什么在call方法中調用onNext(),onCompleted()會觸發subscriber/observer中對應的方法呢？接下來一起看一下源碼，看看是如何訂閱成功的。

進入到Observable.create()方法

public final static <T> Observable<T> create(OnSubscribe<T> f) {return new Observable<T>(hook.onCreate(f));} protected Observable(OnSubscribe<T> f) {this.onSubscribe = f; }

這里創建了一個Observable對象，同時通過構造函數將 f 賦值給Observalbe類中的onSubscribe

進入到subscribe方法

public final Subscription subscribe(Subscriber<? super T> subscriber) {return Observable.subscribe(subscriber, this);}private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {// validate and proceedif (subscriber == null) {throw new IllegalArgumentException("observer can not be null");}if (observable.onSubscribe == null) {throw new IllegalStateException("onSubscribe function can not be null.");/** the subscribe function can also be overridden but generally that's not the appropriate approach* so I won't mention that in the exception*/}// new Subscriber so onStart itsubscriber.onStart();/** See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls* to user code from within an Observer"*/// if not already wrappedif (!(subscriber instanceof SafeSubscriber)) {// assign to `observer` so we return the protected versionsubscriber = new SafeSubscriber<T>(subscriber);}// The code below is exactly the same an unsafeSubscribe but not used because it would // add a significant depth to already huge call stacks.try {// allow the hook to intercept and/or decoratehook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);return hook.onSubscribeReturn(subscriber);} catch (Throwable e) {// special handling for certain Throwable/Error/Exception typesExceptions.throwIfFatal(e);// if an unhandled error occurs executing the onSubscribe we will propagate ittry {subscriber.onError(hook.onSubscribeError(e));} catch (Throwable e2) {Exceptions.throwIfFatal(e2);// if this happens it means the onError itself failed (perhaps an invalid function implementation)// so we are unable to propagate the error correctly and will just throwRuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);// TODO could the hook be the cause of the error in the on error handling.hook.onSubscribeError(r);// TODO why aren't we throwing the hook's return value.throw r;}return Subscriptions.unsubscribed();}}

我們看到

hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);

其中hook.onSubscribeStart返回的是create()方法中創建的OnSubscribe對象，即上文中提到的 onSubscribe ,在這里調用OnSubscribe 中的call方法，將subscriber/observer傳遞到call方法中，所以我們在call中調用onNext(),onError(),onCompleted()會觸發observer中對應的方法，從而達到了事件通知的效果。

2.訂閱Action

觀察者除了我們的subscriber/observer之外，還可以是Action

再看一個例子
定義一個action

private Action1<String> action1 = new Action1<String>() {@Overridepublic void call(String s) {Log.e("HP",s);} }; observable.subscribe(action1);

再看輸出結果

test1 test2

不出所料，輸出結果與預期的相同。
進入subscribe方法看以看究竟

public final Subscription subscribe(final Action1<? super T> onNext) {if (onNext == null) {throw new IllegalArgumentException("onNext can not be null");}return subscribe(new Subscriber<T>() {@Overridepublic final void onCompleted() {// do nothing}@Overridepublic final void onError(Throwable e) {throw new OnErrorNotImplementedException(e);}@Overridepublic final void onNext(T args) {onNext.call(args);}});

原來在subscribe方法中將我們的action轉換成了subscribe。
所以通過action你可以自定義一些實現onNext(),onError(),onComplete()方法的action，這些action通常是可以作為公用的操作。

這些僅僅是RxJava最基礎用法的一個解析。

總結

以上是生活随笔為你收集整理的RxJava初级解析（一）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。