Observer观察者与Observable可观察对象在Rxja

作者: coke613 | 来源:发表于2019-08-09 10:48 被阅读0次

既然选择了远方,便只顾风雨兼程.

该图片来自网络资源,若有侵权,请留言,我将自行删除

要说Android界网络请求框架,那就不得不提Rxjava + Retrofit.
Retrofit,是目前主流网络请求框架,功能强大,方便操作.
Rxjava,异步实现操作的库,可在线程间快速切换，并提供许多操作符,将复杂臃肿代码变得清晰易懂。
在Rxjava中通常表现为Observer(观察者)订阅Observable(可观察对象),通过创建可观察对象发送的数据流.在此过程中,可通过操作符,线程切换等操作最后由观察者接受并做出响应的一个过程.

Observable

Observable 具体表现为一个抽象类,实现ObservableSource接口

public abstract class Observable<T> implements ObservableSource<T> {
      //......
}

ObservableSource 接口

public interface ObservableSource<T> {

    /**
     * Subscribes the given Observer to this ObservableSource instance.
     * @param observer the Observer, not null
     * @throws NullPointerException if {@code observer} is null
     */
    void subscribe(@NonNull Observer<? super T> observer);
}

由此可见Observable是ObservableSource接口下的一个抽象类.可通过ObservableOnSubscribe 创建可观察对象发射数据流。

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
      @Override
      public void subscribe(ObservableEmitter<String> emitter) throws Exception {
             emitter.onNext("hello world");     
             emitter.onComplete();
      }
 });

调用Observable.create方法，创建一个可观察对象，并通过onNext发送一条数据“hello world”，然后通过onComplete发送完成通知。

Observer

创建观察者,接收可观察对象发送的数据流

Observer<String> observer = new Observer<String>() {
       @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(String s) {
            LogUtils.e(s);
        }

        @Override
        public void onError(Throwable e) {
             e.printStackTrace();
         }

        @Override
        public void onComplete() {
             LogUtils.e("------接收完毕------");
         }
  };

关联观察者与可观察者对象

 observable.subscribe(observer);

通过subscribe方法,观察者可订阅被观察者,一旦被订阅后,此时两者为一体.Observer便可对Observable中的行为作出响应。

控制台打印:

E/LogUtils: hello world
E/LogUtils: ------接收完毕------

Emitter / Observer

我们在使用Observable.create方法时,往参数中扔了一个ObservableOnSubscribe实例对象,而真正发射数据流是ObservableEmitter.

public interface ObservableEmitter<T> extends Emitter<T> {
    //.......
}

public interface Emitter<T> {

    /**
     * 用来发送数据,可多次调用,每调用一次发送一次数据
     * Signal a normal value.
     * @param value the value to signal, not null
     */
    void onNext(@NonNull T value);

    /**
     *用来发送异常通知，只发送一次，若多次调用只发送第一条
     * Signal a Throwable exception.
     * @param error the Throwable to signal, not null
     */
    void onError(@NonNull Throwable error);

    /**
     * 用来发送完成通知，只发送一次，若多次调用只发送第一条
     * Signal a completion.
     */
    void onComplete();
}

onNext() 用来发送数据流,可多次调用,每调用一次发送一次数据.
onError() 发送异常通知,只调用一次.多次调用只发送第一条.
onComplete() 发送完成数据流发送通知,只调用一次,多次调用只发送第一条.若数据在全部发送完之后均正常，可以调用onComplete发送一条完成通知
其中onError() 与 onComplete() 为唯一且互斥函数.

接口Observer中的三个方法（onNext,onError,onComplete）正好与Emitter中的三个方法相对应，对于Emitter中对应方法发送的数据或通知进行响应。

优化demo: 省略变量

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("hello world");
                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {    
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                LogUtils.e(s);
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {
                 LogUtils.e("------接收完毕------");
            }
        });

使用修饰符优化demo

 Observable.just("Hello world").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtils.e( s);
            }
        });

 Observable.just("hello-world").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {   // 接收正常发射的数据
                 LogUtils.e(s);
            }
        }, new Consumer<Throwable>() {       //接收可观察者异常信息
            @Override
            public void accept(Throwable throwable) throws Exception {
                throwable.printStackTrace();
            }
        }, new Action() {           // 接收完成信息,作出响应行动。
            @Override
            public void run() throws Exception {
                 LogUtils.e("------接收完毕------");
            }
        });

just 修饰符 : 发送单条数据.(数字,字符,字符串,数组,集合)都可当做单条数据发送.
Consumer 可以看做是对观察者Observer功能单一化之后的产物.
第一个参数Consumer泛型为发射数据类型,通过函数accept进行数据流响应处理.
第二个参数Consumer规定泛型<Throwable>通过函数accept接收异常信息。
第三个参数Action也是对观察者Observer功能单一化之后的产物--行动，通过函数run接收完成信息，作出响应行动。

使用λ表达式

Observable.just("hello world").subscribe(s -> LogUtils.e(s));

发送数据序列

private void demo5(List<String> list) {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                for (int i = 0; i < list.size(); i++) {
                    emitter.onNext(list.get(i) );
                }
                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
               LogUtils.e(s);
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {
                LogUtils.e("------接收完毕------");
            }
        });
    }

在subscribe方法中,遍历List 集合中的数据,通过 emitter.onNext函数将元素发射出去.遍历完成之后,由emitter.onComplete()函数告知发射数据完毕.
在Observer 接收数据中,函数onNext会一直接收emitter.onNext函数发射的数据,如遇到数据异常则函数onError会被回调.数据发射完毕,回调onComplete函数.

优化demo

 private void demo6(List<String> list) {
        Observable.fromIterable(list).subscribe(s -> LogUtils.e(s));
 }

操作符 fromIterable:用来将一个可迭代对象中的元素逐一发送

Disposable

在Observer onSubscribe函数中参数是Disposable.用来切断observer和Observalbe之间的连接，当调用它的disposable()方法时，它就将观察者和被观察者之间的连接切断，从而使Observer收不到事件。确切的说是将Observer切断，不再接受来自被观察者的事件，但是被观察者事件仍在继续执行。

public interface Disposable {
    /**
     * Dispose the resource, the operation should be idempotent.
     */
    void dispose();

    /**
     * Returns true if this resource has been disposed.
     * @return true if this resource has been disposed
     */
    boolean isDisposed();
}

dispose 主动解除订阅.
isDisposed 判断是否解除订阅.true为取消订阅,观察者不在接收数据流.

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i < 10; i++) {
                    emitter.onNext(i );
                    LogUtils.e("发送者:" + i);
                }
                emitter.onComplete();
            }
        }).subscribe(new io.reactivex.Observer<Integer>() {
            private Disposable mDisposable;

            @Override
            public void onSubscribe(Disposable d) {
                this.mDisposable = d;
            }

            @Override
            public void onNext(Integer s) {
                if (s > 4){
                    mDisposable.dispose();   // 取消订阅
                }
                LogUtils.e("接收者: "+s);
            }

            @Override
            public void onError(Throwable e) {
                 e.printStackTrace();
            }

            @Override
            public void onComplete() {
               LogUtils.e("结束---end");
            }
        });
    }

log日志打印

2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 接收者: 0
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 发送者-->:0
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 接收者: 1
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 发送者-->:1
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 接收者: 2
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 发送者-->:2
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 接收者: 3
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:3
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 接收者: 4
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:4
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 接收者: 5
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:5
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:6
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:7
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:8
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:9

在next函数中,当参数 s>4的情况下,mDisposable.dispose();取消订阅.停止接收Observable发送的事件，取消观察者和被观察者之间的联系后，Observable被观察者的事件仍在继续执行。

参考文档:Rxjava2入门教程二：Observable与Observer响应式编程在Rxjava2中的典型实现

Observer观察者与Observable可观察对象在Rxja

既然选择了远方,便只顾风雨兼程.

Observable

Observer

Emitter / Observer

优化demo: 省略变量

使用修饰符优化demo

使用λ表达式

发送数据序列

优化demo

Disposable

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读