RxJava 和 RxAndroid 一 (基礎)

時間  2019-11-25
標籤 rxjava rxandroid 基礎 欄目 Java 简体版
原文   原文鏈接

一、RxJava 項目地址 html

   https://github.com/ReactiveX/RxJavajava

 

二、RxAndroid 項目地址react

   https://github.com/ReactiveX/RxAndroidandroid

 

三、RxJava 和 RxAndroid 的關係git

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

 

四、RxJava和EventBus的區別？數組

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

 

五、RxAndroid的使用方法maven

    compile 'io.reactivex:rxandroid:1.2.0'ide

   

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

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

 

 七、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

 

 八、建立觀察者

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

　　

九、建立被觀察者

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()）。這樣，由被觀察者調用了觀察者的回調方法，就實現了由被觀察者向觀察者的事件傳遞，即觀察者模式。

　　

 十、訂閱

      因爲觀察者能夠由兩種方式被建立，因此訂閱的方式也有兩種

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;
    }
  1. 在subscribe() 中，首先會調用 onStart() 方法，這個方法前文已經介紹了，是可選的。接着會調用 call()方法，咱們已經分析了在call()方法中會調用屢次 onNext() ，最後調用 onCompleted().看到這裏你就會忽然明白原來subscribe() 方法其實至關於依次執行了：onStart() --> onNext（）--> onCompleted（）
  2. 從這也能夠看出，在 RxJava 中，Observable 並非在建立的時候就當即開始發送事件，而是在它被訂閱的時候，即當 subscribe() 方法執行的時候。
  3. Observer 和 Subscriber 具備相同的角色，並且 Observer 在 subscribe() 過程當中最終會被轉換成 Subscriber對象
  4. 將傳入的 Subscriber 做爲 Subscription 返回。這是爲了方便 unsubscribe().   

 

 十一、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 二（操做符的使用）

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程