RxJava2学习

作者: 麻优夜 | 来源:发表于2020-05-29 18:38 被阅读0次

Android Develop——RxJava2(二) RxJa
继续学习rxjava2
Rxjava2 操作符原理（2）
Rxjava2 线程切换（3）
Rxjava2 基本用法（1）
Rxjava2 简析Flowable背压（4）
RxJava2操作符之“Take”
RxJava2 源码分析二
RxJava2之变换操作符篇
1. RxJava2 : 什么是观察者模式

[参考资料]（https://www.jianshu.com/p/0cd258eecf60）
[参考资料]（https://www.jianshu.com/p/464fa025229e）

RxJava2.jpg

        //创建一个上游 Observable：
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });
        //创建一个下游 Observer
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "subscribe");
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "error");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "complete");
            }
        };
        //建立连接
        observable.subscribe(observer);

这个运行结果就是：

12-02 03:37:17.818 4166-4166/zlc.season.rxjava2demo D/TAG: subscribe
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 1
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 2
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 3
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: complete

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
                Log.d(TAG, "emit 4");
                emitter.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable mDisposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "subscribe");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "onNext: " + value);
                i++;
                if (i == 2) {
                    Log.d(TAG, "dispose");
                    mDisposable.dispose();
                    Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "error");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "complete");
            }
        });

运行结果为:

12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: subscribe
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: dispose
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: isDisposed : true
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 3
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit complete
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 4

subscribe()有多个重载的方法:

// 表示下游不关心任何事件,你上游尽管发你的数据去吧, 老子可不管你发什么.
public final Disposable subscribe() {}
// 表示下游只关心onNext事件, 其他的事件我假装没看见
public final Disposable subscribe(Consumer<? super T> onNext) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} 
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
public final void subscribe(Observer<? super T> observer) {}

如果只需要onNext事件可以这么写:

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
                Log.d(TAG, "emit 4");
                emitter.onNext(4);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "onNext: " + integer);
            }
        });

线程调度器：先改变上游发送事件的线程, 让它去子线程中发送事件, 然后再改变下游的线程, 让它去主线程接收事件。

@Override                                                                                       
protected void onCreate(Bundle savedInstanceState) {                                            
    super.onCreate(savedInstanceState);                                                         
    setContentView(R.layout.activity_main);                                                     
                                                                                                
    Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {   
        @Override                                                                               
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {            
            Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());           
            Log.d(TAG, "emit 1");                                                               
            emitter.onNext(1);                                                                  
        }                                                                                       
    });                                                                                         
                                                                                                
    Consumer<Integer> consumer = new Consumer<Integer>() {                                      
        @Override                                                                               
        public void accept(Integer integer) throws Exception {                                  
            Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());              
            Log.d(TAG, "onNext: " + integer);                                                   
        }                                                                                       
    };                                                                                          
                                                                                                
    observable.subscribeOn(Schedulers.newThread())     // subscribeOn() 指定的是上游发送事件的线程                                         
            .observeOn(AndroidSchedulers.mainThread()) // observeOn() 指定的是下游接收事件的线程                                          
            .subscribe(consumer);                                                               
}

在RxJava中, 已经内置了很多线程选项供我们选择, 例如有：

Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
Schedulers.newThread() 代表一个常规的新线程
AndroidSchedulers.mainThread() 代表Android的主线程

注意:

只有当上游和下游建立连接之后, 上游才会开始发送事件. 也就是调用了subscribe() 方法之后才开始发送事件。
上游可以发送无限个onNext, 下游也可以接收无限个onNext。
当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件。
当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件。
上游可以不发送onComplete或onError。
最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然。（比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了, 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃。）
调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件。
上下游默认是在同一个线程工作。也就是说上游在哪个线程发事件, 下游就在哪个线程接收事件。
多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略。多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次。
RxJava中已经内置了一个容器CompositeDisposable,每当我们得到一个Disposable时就调用CompositeDisposable.add()将它添加到容器中, 在退出的时候, 调用CompositeDisposable.clear() 即可切断所有的水管。

规则示意图.png

名词解释

ObservableEmitter：用来发出事件的，发出三种类型的事件：
onNext(T value)、onComplete()、onError(Throwable error)
Disposable：对应于上面的水管的例子, 我们可以把它理解成两根管道之间的一个机关, 当调用它的dispose()方法时, 它就会将两根管道切断, 从而导致下游收不到事件。

Android Develop——RxJava2(二) RxJa
在RxJava2(一)教程中，已经跟着大神们学习了RxJava2的基本使用，现在我们来学习一下RxJava2很强大...
继续学习rxjava2
学习rxjava2好几天了吧，一开始在各个论坛找关于rxjava2的帖子，学习，但是并不是很明白，最后无意点到Se...
Rxjava2 操作符原理（2）
Rxjava2 基本用法（1） Rxjava2 操作符原理（2） Rxjava2 线程切换（3) Rxjava2 ...
Rxjava2 线程切换（3）
Rxjava2 基本用法（1） Rxjava2 操作符原理（2） Rxjava2 线程切换（3) Rxjava2 ...
Rxjava2 基本用法（1）
Rxjava2 基本用法（1） Rxjava2 操作符原理（2） Rxjava2 线程切换（3) Rxjava2 ...
Rxjava2 简析Flowable背压（4）
Rxjava2 基本用法（1） Rxjava2 操作符原理（2） Rxjava2 线程切换（3) Rxjava2 ...
RxJava2操作符之“Take”
前言最近我也在学习RxJava2,在网上找了好多文章来读，发现大多数都是说RxJava2和RxJava之间到底有...
RxJava2 源码分析二
文章目录前言 RxJava2 线程调度 RxJava2 怎么进行线程调度总结前言经过RxJava2源码分析...
RxJava2之变换操作符篇
在上一篇文章中,我们介绍了RxJava2的基本简单实用,在本篇文章中,我们将继续学习RxJava2中的常用操作符 ...
1. RxJava2 : 什么是观察者模式
1. RxJava2 : 什么是观察者模式2. RxJava2 : 创建操作符(无关时间)3. Rxjava2 :...