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教程