當前位置：首頁 > 运维知识 > Android >内容正文

Android

RxJava 和 RxAndroid 一 (基础)

發布時間：2025/5/22 Android 39 豆豆

生活随笔收集整理的這篇文章主要介紹了 RxJava 和 RxAndroid 一 (基础) 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

1、RxJava 項目地址

? ?https://github.com/ReactiveX/RxJava

2、RxAndroid 項目地址

? ?https://github.com/ReactiveX/RxAndroid

3、RxJava 和 RxAndroid 的關系

? ? ?RxAndroid是RxJava的一個針對Android平臺的擴展，主要用于 Android 開發

4、RxJava和EventBus的區別？

? ? ?https://www.zhihu.com/question/32179258/answer/54989242

5、RxAndroid的使用方法

? ??compile 'io.reactivex:rxandroid:1.2.0'

? ?

6、如何查看RxAndroid最新版本？

? ? http://search.maven.org/#search%7Cga%7C1%7Ca%3A%22rxandroid%22

?7、RxAndroid具體使用方法

? ? ?http://gank.io/post/560e15be2dca930e00da1083#toc_14

? ? ?http://blog.csdn.net/theone10211024/article/details/50435325

? ? http://huxian99.github.io/tags/RxJava/

? ??https://github.com/mcxiaoke/RxDocs

?8、創建觀察者

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55

package lib.com.myapplication; import android.support.v7.app.AppCompatActivity; import android.os.Bundle; import rx.Observer; import rx.Subscriber; public class Main2Activity extends AppCompatActivity { ????@Override ????protected void onCreate(Bundle savedInstanceState) { ????????super.onCreate(savedInstanceState); ????????setContentView(R.layout.activity_main2); ????????//創建觀察者 2 種方法 ????????Observer<String> observer = new Observer<String>() { ????????????@Override ????????????public void onCompleted() { ????????????} ????????????@Override ????????????public void onError(Throwable e) { ????????????} ????????????@Override ????????????public void onNext(String s) { ????????????} ????????} ; ????????// Subscriber 繼承 Observer ，對Observer類做了擴展 ????????Subscriber<String> subscriber = new Subscriber<String>() { ????????????@Override ????????????public void onCompleted() { ????????????} ????????????@Override ????????????public void onError(Throwable e) { ????????????} ????????????@Override ????????????public void onNext(String s) { ????????????} ????????} ; ????} }

從上文可以看到，Subscriber繼承Observer, 只是?Subscriber對Observer做了一些擴展。Subscriber的使用和Observer完全一樣。
?Subscriber 多了一個?onStart 方法
onStart(): 這是?Subscriber?增加的方法。它會在 subscribe 剛開始，而事件還未發送之前被調用，可以用于做一些準備工作，例如數據的清零或重置。這是一個可選方法，默認情況下它的實現為空。需要注意的是，如果對準備工作的線程有要求（例如彈出一個顯示進度的對話框，這必須在主線程執行），?onStart()?就不適用了，因為它總是在 subscribe 所發生的線程被調用，而不能指定線程。要在指定的線程來做準備工作，可以使用?doOnSubscribe()?方法，具體可以在后面的文中看到。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

// Subscriber 繼承 Observer ，對Observer類做了擴展 ???????Subscriber<String> subscriber = new Subscriber<String>() { ???????????@Override ???????????public void onCompleted() { ???????????} ???????????@Override ???????????public void onError(Throwable e) { ???????????} ???????????@Override ???????????public void onNext(String s) { ???????????} ???????????@Override ???????????public void onStart() { ???????????????super.onStart(); ???????????} ???????} ;

9、創建被觀察者

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

//create方式 ?Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() { ?????@Override ?????public void call(Subscriber<? super String> subscriber) { ?????????subscriber.onNext( "aa" ) ; ?????????subscriber.onNext( "bb" ) ; ?????????subscriber.onNext( "cc" ) ; ?????????subscriber.onCompleted(); ?????} ?}); ?//just方式? 最多支持10個數據 ?Observable<String> observable1 = Observable.just( "aa" , "bb" , "cc") ; ?// 將會依次調用： ?// onNext("aa"); ?// onNext("bb"); ?// onNext("cc"); ?// onCompleted(); ?//from方式 ?//1:集合 ?List<String> list = new ArrayList<>() ; ?list.add( "aa" ) ; ?list.add( "bb" ) ; ?list.add( "cc" ) ; ?Observable<String> observable2 = Observable.from( list ) ; ?//2：數組 ?String[] words = { "aa", "bb", "cc" }; ?Observable<String> observable3 = Observable.from( words ) ;

Call()方法：當?Observable?被訂閱的時候，OnSubscribe?的?call()?方法會自動被調用，事件序列就會依照設定依次觸發（對于上面的代碼，就是觀察者Subscriber?將會被調用三次?onNext()?和一次?onCompleted()）。這樣，由被觀察者調用了觀察者的回調方法，就實現了由被觀察者向觀察者的事件傳遞，即觀察者模式。

?10、訂閱

? ? ? 由于觀察者可以由兩種方式被創建，所以訂閱的方式也有兩種

1 2	observable.subscribe( observer ) ; observable.subscribe( subscriber ) ;

Observable.subscribe(Subscriber)?的內部實現是這樣的（僅核心代碼）：

1 2 3 4 5 6 7

// 注意：這不是 subscribe() 的源碼，而是將源碼中與性能、兼容性、擴展性有關的代碼剔除后的核心代碼。 // 如果需要看源碼，可以去 RxJava 的 GitHub 倉庫下載。 public Subscription subscribe(Subscriber subscriber) { ????subscriber.onStart(); ????onSubscribe.call(subscriber); ????return subscriber; }

在subscribe() 中，首先會調用?onStart() 方法，這個方法前文已經介紹了，是可選的。接著會調用 call()方法，我們已經分析了在call()方法中會調用多次 onNext() ，最后調用?onCompleted().看到這里你就會突然明白原來subscribe() 方法其實相當于依次執行了：onStart() --> onNext（）--> onCompleted（）

從這也可以看出，在 RxJava 中，Observable?并不是在創建的時候就立即開始發送事件，而是在它被訂閱的時候，即當?subscribe()?方法執行的時候。

Observer?和?Subscriber?具有相同的角色，而且?Observer?在?subscribe()?過程中最終會被轉換成?Subscriber對象

將傳入的?Subscriber?作為?Subscription?返回。這是為了方便?unsubscribe().? ?

?11、RxBus

? ? ? ? ?你是否聽說過EventBus , 他是android 中的事件總線。用rxjava同樣可以實現android的事件總線功能，也就是RxBus.

? ? ? ? 關于rxbus 的基本說明在這里?http://nerds.weddingpartyapp.com/tech/2014/12/24/implementing-an-event-bus-with-rxjava-rxbus/

? ? ? ? 然而這并沒有什么卵用！

? ? ? ? ?下面是RxBus的封裝版 ? ?　 ??

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85

package lib.com.myapplication; import android.support.annotation.NonNull; import android.util.Log; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.concurrent.ConcurrentHashMap; import rx.Observable; import rx.subjects.PublishSubject; import rx.subjects.Subject; /** ?* Created by ${zyj} on 2016/5/6. ?*/ public class RxBus { ????private static final String TAG = RxBus.class.getSimpleName(); ????private static RxBus instance; ????public static boolean DEBUG = false; ????public static RxBus get() { ????????if (instance == null) { ????????????synchronized (RxBus.class) { ????????????????if (instance == null) { ????????????????????instance = new RxBus(); ????????????????} ????????????} ????????} ????????return instance; ????} ????private RxBus() { ????} ????private ConcurrentHashMap<Object, List<Subject>> subjectMapper = new ConcurrentHashMap<>(); ????@SuppressWarnings("unchecked") ????public <T> Observable<T> register(@NonNull Object tag, @NonNull Class<T> clazz) { ????????List<Subject> subjectList = subjectMapper.get(tag); ????????if (null == subjectList) { ????????????subjectList = new ArrayList<>(); ????????????subjectMapper.put(tag, subjectList); ????????} ????????Subject<T, T> subject; ????????subjectList.add(subject = PublishSubject.create()); ????????if (DEBUG) Log.d(TAG, "[register]subjectMapper: " + subjectMapper); ????????return subject; ????} ????public void unregister(@NonNull Object tag, @NonNull Observable observable) { ????????List<Subject> subjects = subjectMapper.get(tag); ????????if (null != subjects) { ????????????if ( observable != null? && subjects.contains( observable )){ ????????????????subjects.remove((Subject) observable); ????????????} ????????????if (isEmpty(subjects)) { ????????????????subjectMapper.remove(tag); ????????????} ????????} ????????if (DEBUG) Log.d(TAG, "[unregister]subjectMapper: " + subjectMapper); ????} ????public void post(@NonNull Object content) { ????????post( content.getClass().getName(), content); ????} ????@SuppressWarnings("unchecked") ????public void post(@NonNull Object tag, @NonNull Object content) { ????????List<Subject> subjectList = subjectMapper.get(tag); ????????if (!isEmpty(subjectList)) { ????????????for (Subject subject : subjectList) { ????????????????subject.onNext(content); ????????????} ????????} ????????if (DEBUG) Log.d(TAG, "[send]subjectMapper: " + subjectMapper); ????} ????private boolean isEmpty(Collection collection) { ????????return null == collection || collection.isEmpty(); ????} }

　　RxBus的使用

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37

package lib.com.myapplication; import android.os.Bundle; import android.support.v7.app.AppCompatActivity; import rx.Observable; import rx.functions.Action1; public class Activity1 extends AppCompatActivity { ????String tag = "tag" ; ????Observable<String> ob ; ????@Override ????protected void onCreate(Bundle savedInstanceState) { ????????super.onCreate(savedInstanceState); ????????setContentView(R.layout.activity1); ????????//創建被觀察者 ????????ob = RxBus.get().register( tag , String.class ) ; ????????//訂閱觀察事件 ????????ob.subscribe(new Action1<String>() { ????????????@Override ????????????public void call(String s) { ????????????????System.out.println( "fff-- " + s? ); ????????????} ????????}) ; ????????//發送內容 ????????RxBus.get().post(? tag , "我是內容" ); ????} ????@Override ????protected void onDestroy() { ????????super.onDestroy(); ????????//取消訂閱 ????????RxBus.get().unregister( tag , ob ); ????} }

在Activity銷毀的時候，要取消訂閱服務。否則 post() 次數會隨著post()調用逐漸增加
除了上面的簡單使用外，還可以使用 ?Schedulers 、AndroidSchedulers 進行線程切換

RxJava 和 RxAndroid 二（操作符的使用）

轉載于:https://www.cnblogs.com/ldq2016/p/6594464.html

總結

以上是生活随笔為你收集整理的RxJava 和 RxAndroid 一 (基础)的全部內容，希望文章能夠幫你解決所遇到的問題。

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

上一篇： CentOS的安装
下一篇：【bzoj2751】[HAOI2012]