观察者 rxjava

RxJava观察者模式

1.RxJava的观察者模式

RxJava的观察者模式是扩展的观察者模式，扩展的地方主要体现在事件通知的方式有很多种

2.RxJava的观察者模式涉及到几个类

Observable：被观察者
Observer：观察者
Subscribe：订阅
Event：被观察者通知观察者的事件

3.Obsercerable与Observer通过Subscribe实现关联，Event主要向Observer通知Observeble的变化，Event有几个通知方式

Next：常规事件，可以传递各种各样的数据
Error：异常事件，当被观察者发送异常事件后那么其他的事件就不会再继续发送了
Completed：结束事件，当观察者接收到这个事件后就不会再接收后续被观察者发送过来的事件

4.代码实现

首先定义一个观察者Observer

public abstract class Observer<T> {

    //和被观察者订阅后，会回调这个方法

    public static void onSubscribe(Emitter emitter);


     // 传递常规事件，用于传递数据

    public abstract void onNext(T t);


    // 传递异常事件

    public abstract void onError(Throwable e);


    // 传递结束事件

    public abstract void onComplete();

}


Observer中的方法都是回调，其中多了一个Emitter的接口类，他是一个发射器


public interface Emitter<T> {


    void onNext(T t);


    void onError(Throwable error);


    void onCompleted();

}


实现逻辑就是通过包装Observer，里面最终是通过Observer进行回调的


public class CreateEmitter<T> implements Emitter<T> {


    final Observer<T> observer;


    CreateEmitter(Observer<T> observer) {

        this.observer = observer;

    }


     @Override

    public void onNext(T t) {

        observer.onNext(t);

    }


    @Override

    public void onError(Throwable error) {

        observer.onError(error);

    }


    @Override

    public void onComplete() {

        observer.onComplete();

    }

}

被观察者的实现

public abstract class Observable<T>{


	public void subscribe(Observer<T> observer) {

        //通过传入的Observer包装成CreateEmitter，用于回调

    	CreateEmitter emitter = new CreateEmitter(observer);


        //回调订阅成功的方法

        observer.onSubscribe(emitter);


        //回调发射器emitter

        subscribe(emitter);

    }


    /**

    *	订阅成功后，进行回调

    */

    public abstract void subscribe(Emitter<T> emitter);

}


就两步，第一步用于订阅，第二步用于回调

具体的使用

private void observer() {

    // 第一步，创建被观察者

    Observable<String> observable = new Observable<String>() {

            @Override

            public void subscribe(Emitter<String> emitter) {

                emitter.onNext("第一次");


                emitter.onNext("第二次");


                emitter.onNext("第三次");


                emitter.onComplete();

            }

        };


    // 第二步，创建观察者

    Observer<String> observer = new Observer<String>() {

            @Override

            public void onSubscribe(Emitter emitter) {

                Log.i("TAG", " onSubscribe ");

            }


            @Override

            public void onNext(String s) {

                Log.i("TAG", " onNext s:" + s);

            }


            @Override

            public void onError(Throwable e) {

                Log.i("TAG", " onError e:" + e.toString());

            }


            @Override

            public void onComplete() {

                Log.i("TAG", " onComplete ");

            }

        };


    // 第三步，被观察者订阅观察者

    observable.subscribe(observer);

}


被订阅成功后，被观察者的subscribe里面就可以通过发射器发送事件了，最终在观察者的方法里进行回调。

RxJava也是观察者和被观察者订阅的过程，只不过被观察者有变化的时候是由发射器进行发送的，这样就不止有一种事件了

1.RxJava的装饰者模式

- 装饰者模式：在不改变原有的架构基础上添加一些新的功能，是作为其原有结构的包装，这个过程称为装饰。
- RxJava的装饰者模式主要是用于实现Observable和Observer的包装，主要是为了与RxJava的观察者模式配合实现代码的方式更简洁。
- 拆解RxJava的装饰器模式

- - 被观察者Observable

参考手机包装的例子

第一步：要有一个抽象接口，在RxJava中这个抽象接口是ObservableSource，里面有一个方法subscribe


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

}


第二步：要有一个包装类，实现了ObservableSource的，RxJava的包装类是Observable，实现了对应的接口，

并且在subscribe方法里通过调用抽象方法subscribeActual，来对观察者进行订阅

public abstract class Observable<T> implements ObservableSource<T> {

	...


    @Override

    public final void subscribe(Observer<? super T> observer) {

       	...

         subscribeActual(observer);

        ...

    }


 	protected abstract void subscribeActual(Observer<? super T> observer);

    ...

}


第三步：这就是具体的包装类了如图所示

2.观察者Observer：

第一步：要有一个抽象接口，而RxJava的接口是Emitter和Observer，里面有好几个方法基本一样，onNext，onError，onComplete，用于被观察者进行回调；
第二步：要有一个包装类，实现了Emitter或者Observer，但是观察者比较特殊，没有一个基础的包装类，而是直接封装了很多的包装类