Rxjava初步理解

1、Rxjava是一个可以实现异步操作的库

他可以使得代码的编写变得更为简洁（不是代码数量的减少，而是可读性提高）

2、Rxjava的实现是通过观察者模式实现的

通过观察者（Observer）订阅事件（subscribe()方法），被观察者（Observable）数据发生变化时，通知观察者。

3、下面分别描述观察者Observer与被观察者Observable和订阅subscribe

3.1、被观察者（Observable）

3.1.1 通过create方法创建observable，在OnSubscribe中可以执行onNext()、onCompleted()、onError()方法

Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("hello");
        subscriber.onNext("world");
        subscriber.onCompleted();
        subscriber.onError(new Throwable("error"));
    }
});

3.2 观察者（Observer）

3.2.1 observer 对应于observable的三种方法，并做相互匹配，observable调用onNext则observer也调用onNext

Observer<String> observer = new Observer<String>() {
    @Override
    public void onCompleted() {
        
    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(String s) {

    }
};

3.2.2 还有一个Subscribe的抽象类，实现了Observer接口，也可以进行事件的接收。

Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onCompleted() {
        
    }

    @Override
    public void onError(Throwable e) {
        
    }

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

3.3 订阅：观察者通过订阅的方式观察被观察者。

observable.subscribe(observer);
observable.subscribe(subscriber);

这个方法的字面理解是被观察者订阅观察者，貌似意思反了，其实是为了方便代码的书写。

3.4 subscribe()源码的简单分析

翻看subscribe()方法的源码，其实很简单，只是进行了三步操作

1、执行subscriber.start()方法,执行准备
2、执行onSubscribe.call(subscriber)
3、返回onSubscribe,方便调用unsubscribe

4、 RxJava 还提供了一些用来快捷创建事件队列的方法

4.1 just(T...)方法

逐一发送事件消息

Observable<String> observable = Observable.just("hello","world","你好");

4.2 from(T[])方法

将消息进行拆分并逐一发送

String[] word = {"hello","world","你好"};
Observable<String> observable = Observable.from(word);

5、 ActionX()方法

RxJava 是提供了多个 ActionX()形式的接口，用于包装不同的方法

Action0 onCompleteAction = new Action0() {
     @Override
     public void call() {
         Log.d(TAG, "complte");
     }
 };

Action1<String> action1 = new Action1<String>() {
     @Override
     public void call(String s) {
         Log.d(TAG, s);
     }
 };
 
 Action1<Throwable> onErrorAction = new Action1<Throwable>() {
     @Override
     public void call(Throwable throwable) {
         Log.d(TAG, "error");
     }
 };

Action2<String, Throwable> action2 = new Action2<String, Throwable>() {
    @Override
    public void call(String s, Throwable throwable) {
        Log.d(TAG, "call: " + s + "," + throwable.getMessage());      
    }
};

由上可知，Action0用于包装onCompleted()这类无返回值类型的方法，Action1<T>用于包装onNext()和onError()这类无返回值类型的方法，Action2<T1,T2>用于包装两种无返回值类型的方法......但是subscribe()方法只能接收Action0()与Action1()这两种。

6、事件的发送与接收

当Observable通过onNext()发送消息的消息是逐条进行处理的，当observable被订阅后，自动执行OnSubscribe中的call()方法。每发送一条数据，Observer接收一条数据，直到消息发送完成或者发送onCompleted()、onError()方法后Observer停止接收。

7、事件的线程Schedule

Rxjava内置了几个Schedule

Schedulers.immediate() 执行在当前的线程中
Schedulers.newThread() 创建新线程并执行
Schedulers.io() 用于IO密集型任务，如异步阻塞IO操作，这个调度器的线程池会根据需要增长
Schedulers.computation() 用于计算任务，如事件循环或和回调处理，不要用于IO操作(IO操作请使用Schedulers.io())；默认线程数等于处理器的数量