一、前言

阅读本文前，建议详细阅读并掌握什么是观察者模式。

https://github.com/ReactiveX/RxJava
https://github.com/ReactiveX/RxAndroid
RxJava是ReactiveX的一种Java的实现形式，拥有三大部分

• 观察者模式，即定义对象间一种一对多的依赖关系，当一个对象改变状态时，则所有依赖它的对象都会被改变。
• Iterator模式，即迭代流式编程模式。
• 函数式编程模式，即提供一系列函数样式的方法供快速开发。

首先RxAndroid基于RxJava，RxAndroid结合Android添加了很少的类，让我们在Android更方便和简单的使用RxJava。RxJava和RxAndroid已经来到了3.0+版本

RxJava核心思想：观察者模式 + 异步 来处理事件

implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
implementation 'io.reactivex.rxjava3:rxjava:3.0.0'

二、基本用法

        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("one");
                emitter.onNext("two");
                //emitter.onComplete();
                emitter.onError(new Throwable("error"));
            }
        });
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.d("yink","onSubscribe ...");
            }

            @Override
            public void onNext(@NonNull String s) {
                Log.d("yink","onNext = " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.d("yink","onError ... = " + e);
            }

            @Override
            public void onComplete() {
                Log.d("yink","onComplete ...");
            }
        };
        observable.subscribeOn(Schedulers.newThread());
        observable.observeOn(AndroidSchedulers.mainThread());
        Log.d("yink","subscribe ...");
        observable.subscribe(observer);

首先，RxJava基于观察者模式，所以它的结构很简单，无非观察者模式三步骤：

• Observable被观察者，ObservableEmitter为被观察者的几个事件。（onComplete和onError唯一且互斥）
• Observer观察者，观察几个方法
• observable.subscribe(observer)提交观察者到被观察者

其次，RxJava还额外实现了线程调度，即控制观察者和被观察者的代码在哪个线程执行：

• observable.subscribeOn(Schedulers.newThread());设置被观察者开启新线程来调度
• observable.observeOn(AndroidSchedulers.mainThread());设置观察者在主线程调度
  所以被观察者中我们就可以做一些耗时操作，然后ObservableEmitter通知到观察者，观察者中可以刷新UI等操作。

2020-03-18 10:36:27.841 12678-12678/com.example.demo D/yink: subscribe ...
2020-03-18 10:36:27.841 12678-12678/com.example.demo D/yink: onSubscribe ...
2020-03-18 10:36:27.841 12678-12678/com.example.demo D/yink: onNext = one
2020-03-18 10:36:27.841 12678-12678/com.example.demo D/yink: onNext = two
2020-03-18 10:36:27.841 12678-12678/com.example.demo D/yink: onError ... = java.lang.Throwable: error

整体看来RxJava是为了让我们更加优雅的异步，异步指定线程方便，过程好控制，还可多个观察者，指定不同线程，这样异步真的太方便了。

三、详细介绍

现在我们知道RxJava大体框架，观察者模式+异步。下面我们来看详细介绍。详细介绍分为下面几个部分：

• Observer 观察者的各种扩展写法
• Observable 被观察者的各种拓展写法
• 线程调度，指定线程操作相关

3.1、观察者Observer

观察者的扩展写法就比较简单

public final Disposable subscribe() 
void subscribe(@NonNull Observer<? super T> observer);
public final Disposable subscribe(@NonNull Consumer<? super T> onNext) 
public final Disposable subscribe(@NonNull Consumer<? super T> onNext, 
      @NonNull Consumer<? super Throwable> onError) 
public final Disposable subscribe(@NonNull Consumer<? super T> onNext, 
      @NonNull Consumer<? super Throwable> onError,
      @NonNull Action onComplete) 
public final Disposable subscribe(@NonNull Action onComplete) {
public final Completable doOnSubscribe(@NonNull Consumer<? super Disposable> onSubscribe) {
public final void blockingSubscribe(@NonNull Action onComplete) {
...

• 可以看到我们提交观察者时，观察者的几个变种
• subscribe() 实际提交的是一个Functions.emptyConsumer()
• onNext事件：Consumer<? super T>
• onError事件：Consumer<? super Throwable>
• onComplete事件：Action
• Observer<? super T> observer:多种继承自Observer的类也可以
  所以，我们在监听上，就可以简单只监听部分事件。例如

Observable.just("one").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {

            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
                
            }
        });

这个例子就提交了两个观察者。一个观察者只监听onNext事件，一个观察者只监听onError事件。当然你也可以只提交一个观察者。源码如下，很好理解,我随便点了一个subscribe的实现

public final Disposable subscribe(@NonNull Consumer<? super T> onNext, @NonNull Consumer<? super Throwable> onError,
           ...
        LambdaObserver<T> ls = new LambdaObserver<>(onNext, onError, onComplete, Functions.emptyConsumer());
        subscribe(ls);
        return ls;
    }

public LambdaObserver(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
            Action onComplete,
            Consumer<? super Disposable> onSubscribe) {
        super();
        this.onNext = onNext;
        this.onError = onError;
        this.onComplete = onComplete;
        this.onSubscribe = onSubscribe;
    }

@Override
    public void onNext(T t) {
        if (!isDisposed()) {
            try {
                onNext.accept(t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                get().dispose();
                onError(e);
            }
        }
    }

3.2、被观察者Observable

Observable是被观察者产生的地方，分下面几个方面介绍：

• Observable的创建
• Observable操作符归类
• Observable扩展：Flowable、Maybe、Single、Completable

3.2.1、Observable的创建

事件产生理解上很简单，就是创建被观察者,我们以just的为例，看看它的整体思路

Observable.just("a","b").subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.d("yink","onSubscribe ...");
            }

            @Override
            public void onNext(@NonNull String s) {
                Log.d("yink","onNext = " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.d("yink","onError ... = " + e);
            }

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

2020-03-24 09:30:13.562 31403-31403/com.example.demo D/yink: onSubscribe ...
2020-03-24 09:30:13.562 31403-31403/com.example.demo D/yink: onNext = a
2020-03-24 09:30:13.562 31403-31403/com.example.demo D/yink: onNext = b
2020-03-24 09:30:13.562 31403-31403/com.example.demo D/yink: onComplete ...

将对象或者对象集合转换为一个会发射这些对象的Observable
just用法很简单，直接把数组以onNext事件传递，传递完后 调用onComplete事件
我们接着来看看just方法的实现,篇幅原因，我省去了一些代码。

 public static <T> Observable<T> just(@NonNull T item1, @NonNull T item2) {
        ...
        return fromArray(item1, item2);
    }

 public static <T> Observable<T> fromArray(@NonNull T... items) {
        ...
        return RxJavaPlugins.onAssembly(new ObservableFromArray<>(items));
    }

public final class ObservableFromArray<T> extends Observable<T> {
    final T[] array;
    public ObservableFromArray(T[] array) {
        this.array = array;
    }

    @Override
    public void subscribeActual(Observer<? super T> observer) {
        FromArrayDisposable<T> d = new FromArrayDisposable<>(observer, array);

        observer.onSubscribe(d);

        if (d.fusionMode) {
            return;
        }

        d.run();
    }

    static final class FromArrayDisposable<T> extends BasicQueueDisposable<T> {
        ....
        FromArrayDisposable(Observer<? super T> actual, T[] array) {
            this.downstream = actual;
            this.array = array;
        }
        ...
        void run() {
            T[] a = array;
            int n = a.length;

            for (int i = 0; i < n && !isDisposed(); i++) {
                T value = a[i];
                if (value == null) {
                    downstream.onError(new NullPointerException("The element at index " + i + " is null"));
                    return;
                }
                downstream.onNext(value);
            }
            if (!isDisposed()) {
                downstream.onComplete();
            }
        }
    }
}

• 可以看到调用流程也很简单，把数组挨个调用就可以了，最后发出onComplete事件
• 事件的产生代码都极其类似，最后定位实现都会定位到类似ObservableFromArray.java这样的实现
• 事件产生实现的代码路径在“/RxJava-3.x/src/main/java/io/reactivex/rxjava3/internal/operators/observable/**”，此路径下有很多种实现，包括from系列，just，create，error，timer等等，只是RxJava源码为我们默认添加的一些实现。
• 创建观察者思想都类似，就不一一详述了，简单介绍如下：

用法	简单介绍
create	通过调用观察者的方法从头创建一个Observable
frome	封装Iterable、Array、Callable、Action、Runnable、Future，以执行为核心,调用封装的类型的内部事件。将其它的对象或数据结构转换为Observable
targetType.from{sourceType}()	这几种类型都是为了解决特定的问题而存在，3.1.3会简单介绍
just	将对象或者对象集合转换为一个会发射这些对象的Observable
defer	普通observable创建对象时就确定了参数，defer则是我们提交观察者时才去创建被观察者和参数，相当于一个懒加载，在观察者订阅之前不创建这个Observable，为每一个观察者创建一个新的Observable
range	rang(0,10)直接生成0-10事件，创建发射指定范围的整数序列的Observable
interval	interval(1, TimeUnit.SECONDS);周期性生成一个无限的、永远增长的数（长整型）
timer	Observable.timer(5, TimeUnit.MINUTES)延时五分钟发送事件
empty	创建一个不发射任何数据但是正常终止的Observable，只调用一个onCompleted方法
never	创建一个什么事件都不发送的被观察者
error	发送error事件

3.2.2、ObservableObservable操作符归类

Rxjava提供的操作符真的蛮多的。主要分为下面几个大类

• 直接创建一个Observable（创建操作）
• 组合多个Observable（组合操作）
• 对Observable发射的数据执行变换操作（变换操作）
• 从Observable发射的数据中取特定的值（过滤操作）
• 转发Observable的部分值（条件/布尔/过滤操作）
• 对Observable发射的数据序列求值（算术/聚合操作）

这段归类描述引用自知乎的一个回答，Rxjava、rxandroid中的操作，这里我将其制成表格，供大家查阅使用，上面有创建操作符表格，所以下面是省略这部分。

变换操作

过滤操作

组合操作

错误处理
这些操作符用于从错误通知中恢复

辅助操作

条件和布尔操作

算术和聚合操作

连接操作

转换操作

这几个表格的目的是方便大家查阅，我们不需要一口气学习所有操作符，知其思想即可。这些操作符目的加工Observable
由于篇幅原因这里只举一个map例子，上面所有操作符都很容易上手使用，就不过多阐述。

Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onNext(3);
        }
    }).map(new Function<Integer, String>() {
        @Override
        public String apply(Integer integer) throws Exception {
            return "map add " + integer;
        }
    }).subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            Log.d("yink", s);
        }
    });

2020-03-24 14:59:30.518 8467-8467/com.example.demo D/yink: map add 1
2020-03-24 14:59:30.518 8467-8467/com.example.demo D/yink: map add 2
2020-03-24 14:59:30.518 8467-8467/com.example.demo D/yink: map add 3

自定义操作符
自定义操作符有两个方向:

• 对Observable被观察者中的数据动刀，让数据转换一次，最后再发送到观察者。
• 对Observable被观察者直接动刀，转换一次被观察者
  1、实现ObservableOperator接口，转换数据

@FunctionalInterface
public interface ObservableOperator<@NonNull Downstream, @NonNull Upstream> {
    @NonNull
    Observer<? super Upstream> apply(@NonNull Observer<? super Downstream> observer) throws Throwable;
}

Observable.just("1","2")
                  .lift(new YourClass<String>())

2、实现ObservableOperator

public interface ObservableTransformer<Upstream, Downstream> {
    @NonNull
    ObservableSource<Downstream> apply(@NonNull Observable<Upstream> upstream);
}

Observable.just("1","2")
                  .compose(new YourTransformer())
                  .subscribe...

3.2.3、Observable扩展

Flowable、Maybe、Single、Completable

Flowable
Flowable为了解决背压的问题而存在。直接上例子：

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                for(int j = 0; j <= 100; j++){
                    e.onNext(j);
                    Log.i("yink"," send id = " + j);
                    try{
                        Thread.sleep(50);
                    }catch (Exception ex){
                    }
                }
            }
        }, BackpressureStrategy.MISSING)
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(Long.MAX_VALUE); //观察者设置接收事件的数量,如果不设置接收不到事件
                    }
                    @Override
                    public void onNext(Integer integer) {
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        Log.e("yink","onNext = " + integer);
                    }
                    @Override
                    public void onError(Throwable t) {
                        Log.e("yink","onError = " + t.toString());
                    }
                    @Override
                    public void onComplete() {
                        Log.e("yink","onComplete");
                    }
                });

• 背压：被观察者发送事件的速度大于观察者接收的速度时，观察者内部是创建一个无限大的缓存来暂存没有处理的事件。如果事件太多会导致OOM。
• Flowable提供了几种方式来避免无限放大的缓存
• Flowable 在使用上并无太大差别，多了一个BackpressureStrategy参数，必须指定接收事件数量
• Flowable 缓存池默认大小是128
• BackpressureStrategy参数就是Flowable设置在背压发生时处理的方式,参数类型如下：

public enum BackpressureStrategy {
    MISSING,
    ERROR,
    BUFFER,
    DROP,
    LATEST
}

Single
Single，用法如其名，它只有onSuccess和onError事件，只发送一次事件。例子如下比较简单：

      Single.create(new SingleOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull SingleEmitter<String> e) throws Exception {

                e.onSuccess("1");
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.d("yink","s = " + s.toString());
            }
        });

public interface SingleEmitter<@NonNull T> {
    void onSuccess(@NonNull T t);
    void onError(@NonNull Throwable t);
    void setDisposable(@Nullable Disposable d);
    void setCancellable(@Nullable Cancellable c);
    boolean isDisposed();
    boolean tryOnError(@NonNull Throwable t);
}

Completable
Completable不发送任何数据,发送onComplete和onError，用法也比较简单，下面例子就是，线程执行完了，接着可以做别的。因为线程执行时也不用发出什么事件，或者发送值。
Completable还支持toXXX,转换成Flowable/Single/Maybe/Observable

Completable.create(new CompletableOnSubscribe() {
            @Override
            public void subscribe(@NonNull CompletableEmitter emitter) throws Exception {
                try{
                        Thread.sleep(50);
                    }catch (Exception ex){
                    }
            }
        }).andThen(Observable.range(1, 10))
        .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                Log.d("yink","integer = " + integer);
            }
        });

public interface CompletableEmitter {
    void onComplete();
    void onError(@NonNull Throwable t);
    void setDisposable(@Nullable Disposable d);
    void setCancellable(@Nullable Cancellable c);
    boolean isDisposed();
}

Maybe
Maybe可以理解为Single和Completable的结合。看事件就理解了，Maybe只能传一次值onSuccess，然后支持onComplete和onError事件。

      Maybe.create(new MaybeOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull MaybeEmitter<String> e) throws Exception {
                e.onSuccess("testA");
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.d("yink", " s = " + s);
            }
        });

public interface MaybeEmitter<@NonNull T> {
    void onSuccess(@NonNull T t);
    void onError(@NonNull Throwable t);
    void onComplete();
    void setDisposable(@Nullable Disposable d);
    void setCancellable(@Nullable Cancellable c);
    boolean isDisposed();
    boolean tryOnError(@NonNull Throwable t);
}

所以Rxjava提供五种被观察者写法：Observable、Flowable、Maybe、Single、Completable。可以方便我们再合适的地方用合适的类，写出更优美的代码。

3.3、线程调度

此小结为RxJava最骚的地方，正是因为RxJava在观察者模式上增加了线程调度，所以让异步变得如此优美。

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                Log.e("yink", "Observable call: " + Thread.currentThread().getName());
                emitter.onNext("one");
                emitter.onNext("two");
            }
        })
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                Log.e("yink", "Observer call: " + Thread.currentThread().getName());
                Log.d("yink","s = " + s);
            }
        });

2020-03-25 09:31:38.388 29612-29724/com.example.demo E/yink: Observable call: RxNewThreadScheduler-1
2020-03-25 09:31:38.417 29612-29612/com.example.demo E/yink: Observer call: main
2020-03-25 09:31:38.417 29612-29612/com.example.demo D/yink: s = one
2020-03-25 09:31:38.417 29612-29612/com.example.demo E/yink: Observer call: main
2020-03-25 09:31:38.417 29612-29612/com.example.demo D/yink: s = two

RxJava/RxAndroid默认给我们线程调度方式基本涵盖了我们绝大部分会使用的情况。

四、RxAndroid

RxAndroid中的代码很少，它扩展的意义在于方便我们结合Android特性，在UI线程中调度。或者结合Handler调度。举两个例子

public class ReactiveFragment extends Fragment {//在UI线程中的例子
    @Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        Observable.just("one", "two", "three", "four", "five")
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(/* an Observer */);
    }

new Thread(new Runnable() {//在其他线程中的例子
    @Override
    public void run() {
        final Handler handler = new Handler(); //绑定到这个线程的Handler
        Observable.just("one", "two", "three", "four", "five")
                .subscribeOn(Schedulers.newThread())
                .observeOn(HandlerScheduler.from(handler))
                .subscribe(/* an Observer */)
    }
}, "custom-thread-1").start();

五、写在最后

到此RxJava和RxAndroid介绍的差不多了。本文主要目的是全方位的介绍一下RxJava，主要拆分成：被观察者 + 观察者 + 异步，分部分来讲解。来达到对RxJava有一个整体上，不光是用法，还有思想上的理解。希望本文对你有所帮助。有啥对或者不对的地方，欢迎交流，指正。