## 1 简介

函数式编程是一种编程范式，在函数式编程中，函数为一等公民，可以作为另一个函数的参数或者作为返回值。在 Android 中 我们常用的就是 RxJava，目前的版本的就是 RxJava3。Android 编程中 使用 RxJava 主要为解决异步操作越来越复杂的问题，虽然 Android 提供了 AsyncTask、Handler 等来进行异步操作，但是面对复杂逻辑操作，可能节就力不从心。

## 2 基本用法

RxJava 原理就是通过扩展的**观察者模式**来实现，Observable  [əbˈzɜːrvəbl]  被观察者，Obsever  [əbˈzɜːrvər]  观察者。 

RxJava3 中主要有以下**被观察者**：

* Observable：发送 0 或 N 个数据，不支持背压（上游数据发射太快，下游告知上游不要快呀）。
* Flowable：发送 0 或 N 个数据，支持背压。
* Completable：不发射数据，只处理 onComplete 和 onError 事件。
* Single：发送单个数据或一个错误，只处理 onSuccess 和 onError 事件。
* Maybe：发送 0 或 N 个数据。

使用：

```java
//1.创建被观察者（数据源）
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("序列1");
                emitter.onNext("序列2");
                emitter.onComplete();
            }
        });

//2.创建观察者
Observer<String> observer = new Observer<String>() {

        @Override
        public void onSubscribe(@NonNull Disposable d) {//订阅时调用

        }

        @Override
        public void onNext(@NonNull String s) {//将数据加入到事件队列

        }

        @Override
        public void onError(@NonNull Throwable e) {//事件队列异常

        }

        @Override
        public void onComplete() {//事件队列完成

        }
};


//3.订阅（数据源被订阅了）
observable.subscribe(observer);
```

## 3 Subject 和 Processor

Subject 既是 Observable，又是 Observer，可以看做 Observable 和  Observer 之间桥梁。

Processor 和 Subject 的功能几乎是一样，区别在于 Subject 不支持背压，而 Processor 支持背压。

Subject 有 4 种类型：AsyncSubject、BehaviorSubject、ReplaySubject、PublishSubject。

* AsyncSubject：Observer 会接收 AsyncSubject 发送的在 onComplete() 之前的最后一个数据。
* BehaviorSubject：Observer 会先接收 BehaviorSubject 被订阅之前的最后一个事件，然后接收订阅之后发送的所有事件。如果 BehaviorSubject 被订阅之前没有发送任何数据，可以通过BehaviorSubject.createDefault 发送一个默认的事件。
* ReplaySubject：Observer 会接收 ReplaySubject 发送的所有事件，无论 subscribe 订阅是何时开始的。
* PublishSubject：Observer 只接收 PublishSubject 被订阅之后发送的事件。

## 4 操作符

* 创建操作符
  * interval：固定时间间隔发射数据，类似于定时器，Observable.interval(3, TimeUnit.SECONDS)
  * range：发射指定范围的数据，Observable.range(0, 5)
  * repeat：重复发射序列，Observable.range(1, 5).repeat(5)
* 变换操作符
  * map：通过 Function 将源 Observable 转换为新的 Observable 对象并发射
  * flatMap、cast：将 Observable 发射的数据集合变换为 Observables 集合，然后发射，不保证发射顺序，cast 是强制将 Observable 发射的所有数据类型转换为指定类参考数据
  * concatMap：和 flatMap 类似，不过保证发射顺序
  * flatMapIterable：数据包装成 Iterable
  * buffer：将源 Observable 转换为新的 Observable 对象，整个源一起发射
  * groupBy：分组然后发射
* 过滤操作符
  * filter：按照自定义规则过滤数据源
  * elementAt：返回指定位置的数据
  * distinct：去重数据
  * skip：过滤前 n 项数据
  * take：取前 n 项数据
  * ignoreElement：忽略数据，只调用 onComplete 和 onError 
  * throttleFirst：定期发射一个时间段源发射的第一个数据
  * throttlWithTimeOut：通过时间来限流，源 Observable 每次发射出来一个数据后就会进行计时
* 组合操作符
  * merge：多个源合并到一个源发射
  * concat：多个源合并发射，并且按照顺序
  * zip：合并一个或多个源，并指定一个新值
  * combineLastest：最新的数据源和其他源打包发射
* 辅助操作符
  * delay：暂停发射
  * doXX：发射数据监听回调，doOnEach、doOnNext、doOnSubscribe、doOnError、doOnTerminate、finallyDo
  * subscribeOn、observeOn：切换线程
  * timeout：指定一段时长还没发射数据，onError 通知
* 转换操作符
  * toList：组合成 List 发射
  * toSortedList：排序，组合成 List 发射，默认为升序
  * toMap：组合成 Map 发射