RxJava2学习

矫情吗；* 2022-06-01 07:15 242阅读 0赞

**目录**

一、简单使用

二、线程调度

2.1 subScribeOn

2.2 observeOn

2.3 线程切换需要注意

三、操作符

3.1 map

3.2 concat

3.3 FlatMap

3.4 concatMap

3.5 zip

3.6 interval

3.7 distinct

3.8 Filter

3.9 buffer

3.10 timer

3.11 doOnNext

3.12 skip

3.13 take

3.14 just

3.15 Single

3.16 debounce

3.17 defer

3.18 last

3.19 merge

3.20 reduce

3.21 scan

3.22 window

3.23 takeUtil

3.24 takeWhile

3.25  skipUntil

四、Subject

五、Transformer与compose

--------------------

# 一、简单使用 #

两种观察者模式：

*  Observable ( 被观察者 ) / Observer ( 观察者 )
 *  Flowable （被观察者）/ Subscriber （观察者）

![70][]

RxJava 的三部曲：

1.  第一步：初始化 Observable
2.  第二步：初始化 Observer
3.  第三步：建立订阅关系

Disposable 相当于 RxJava 1.x 中的 Subscription， 用于解除订阅。可以看到示例代码中，在 i 自增到 2 的时候，订阅关系被切断。

# 二、线程调度 #

## 2.1 subScribeOn ##

`subscribeOn` 用于指定 `subscribe()` 时所发生的线程。

## 2.2 observeOn ##

`observeOn` 方法用于指定下游 `Observer` 回调发生的线程。

## 2.3 线程切换需要注意 ##

RxJava 内置的线程调度器的确可以让我们的线程切换得心应手，但其中也有些需要注意的地方。

*  简单地说，`subscribeOn()` 指定的就是发射事件的线程，`observerOn` 指定的就是订阅者接收事件的线程。
 *  多次指定发射事件的线程只有第一次指定的有效，也就是说多次调用 `subscribeOn()` 只有第一次的有效，其余的会被忽略。
 *  但多次指定订阅者接收线程是可以的，也就是说每调用一次 `observerOn()`，下游的线程就会切换一次。

修改一下上面的例子：

Observable.create(new ObservableOnSubscribe<Integer>() {
                // 第一步：初始化Observable
                @Override
                public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                    Log.e(TAG, "subscribe thread " + Thread.currentThread().getName());
                    Log.e(TAG, "Observable emit 1" + "\n");
                    e.onNext(1);
                    Log.e(TAG, "Observable emit 2" + "\n");
                    e.onNext(2);
                    Log.e(TAG, "Observable emit 3" + "\n");
                    e.onNext(3);
                    e.onComplete();
                    Log.e(TAG, "Observable emit 4" + "\n");
                    e.onNext(4);
                }
            }).subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnNext(new Consumer<Integer>() {
                        @Override
                        public void accept(Integer integer) throws Exception {
                            Log.e(TAG, "accept thread " + Thread.currentThread().getName());
                            Log.e(TAG, "accept " + integer);
                        }
                    })
                    .observeOn(Schedulers.io())
                    .subscribe(new Observer<Integer>() {
                        // 第三步：订阅
                        // 第二步：初始化Observer
                        private int i;
                        private Disposable mDisposable;
    
                        @Override
                        public void onSubscribe(@NonNull Disposable d) {
                            Log.e(TAG, "onSubscribe thread " + Thread.currentThread().getName());
                            mDisposable = d;
                        }
    
                        @Override
                        public void onNext(@NonNull Integer integer) {
                            Log.e(TAG, "onNext thread " + Thread.currentThread().getName());
                            Log.e(TAG, "onNext " + integer);
                            i++;
                            if (i == 2) { // 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
                                mDisposable.dispose();
                            }
                        }
    
                        @Override
                        public void onError(@NonNull Throwable e) {
                            Log.e(TAG, "onError thread " + Thread.currentThread().getName());
                            Log.e(TAG, "onError : value : " + e.getMessage() + "\n");
                        }
    
                        @Override
                        public void onComplete() {
                            Log.e(TAG, "onComplete thread " + Thread.currentThread().getName());
                            Log.e(TAG, "onComplete" + "\n");
                        }
                    });

输出：

10-16 15:44:44.921 32480-32480/com.hy.rxjava E/TAG: onSubscribe thread main
    10-16 15:44:44.924 32480-32507/com.hy.rxjava E/TAG: subscribe thread RxCachedThreadScheduler-1
    10-16 15:44:44.924 32480-32507/com.hy.rxjava E/TAG: Observable emit 1
    10-16 15:44:44.924 32480-32507/com.hy.rxjava E/TAG: Observable emit 2
    10-16 15:44:44.924 32480-32507/com.hy.rxjava E/TAG: Observable emit 3
    10-16 15:44:44.924 32480-32507/com.hy.rxjava E/TAG: Observable emit 4
    10-16 15:44:44.972 32480-32480/com.hy.rxjava E/TAG: accept thread main
    10-16 15:44:44.972 32480-32480/com.hy.rxjava E/TAG: accept 1
    10-16 15:44:44.972 32480-32480/com.hy.rxjava E/TAG: accept thread main
    10-16 15:44:44.972 32480-32480/com.hy.rxjava E/TAG: accept 2
    10-16 15:44:44.972 32480-32480/com.hy.rxjava E/TAG: accept thread main
    10-16 15:44:44.972 32480-32480/com.hy.rxjava E/TAG: accept 3
    10-16 15:44:44.973 32480-32525/com.hy.rxjava E/TAG: onNext thread RxCachedThreadScheduler-2
    10-16 15:44:44.973 32480-32525/com.hy.rxjava E/TAG: onNext 1
    10-16 15:44:44.973 32480-32525/com.hy.rxjava E/TAG: onNext thread RxCachedThreadScheduler-2
    10-16 15:44:44.973 32480-32525/com.hy.rxjava E/TAG: onNext 2

#  三、操作符 #

## 3.1 map ##

`map` 操作符可以将一个 `Observable` 对象通过某种关系转换为另一个`Observable` 对象。

## 3.2 concat ##

![610][]

`concat` 可以做到不交错的发射两个甚至多个 Observable 的发射事件，并且只有前一个 `Observable` 终止(`onComplete`) 后才会订阅下一个 `Observable`。

采用 concat 操作符先读取缓存再通过网络请求获取数据

想必在实际应用中，很多时候（对数据操作不敏感时）都需要我们先读取缓存的数据，如果缓存没有数据，再通过网络请求获取，随后在主线程更新我们的UI。

`concat` 操作符简直就是为我们这种需求量身定做。

利用 `concat` 的必须调用 `onComplete` 后才能订阅下一个 `Observable` 的特性，我们就可以先读取缓存数据，倘若获取到的缓存数据不是我们想要的，再调用 `onComplete()` 以执行获取网络数据的 `Observable`，如果缓存数据能应我们所需，则直接调用 `onNext()`，防止过度的网络请求，浪费用户的流量。

Observable<FoodList> cache = Observable.create(new ObservableOnSubscribe<FoodList>() {
                @Override
                public void subscribe(@NonNull ObservableEmitter<FoodList> e) throws Exception {
                    Log.e(TAG, "create当前线程:" + Thread.currentThread().getName());
                    FoodList data = CacheManager.getInstance().getFoodListData(); // 在操作符 concat 中，只有调用 onComplete 之后才会执行下一个 Observable
                    if (data != null) { // 如果缓存数据不为空，则直接读取缓存数据，而不读取网络数据
                        isFromNet = false;
                        Log.e(TAG, "\nsubscribe: 读取缓存数据:");
                        runOnUiThread(new Runnable() {
                            @Override
                            public void run() {
                                mRxOperatorsText.append("\nsubscribe: 读取缓存数据:\n");
                            }
                        });
                        e.onNext(data);
                    } else {
                        isFromNet = true;
                        runOnUiThread(new Runnable() {
                            @Override
                            public void run() {
                                mRxOperatorsText.append("\nsubscribe: 读取网络数据:\n");
                            }
                        });
                        Log.e(TAG, "\nsubscribe: 读取网络数据:");
                        e.onComplete();
                    }
                }
            });
            Observable<FoodList> network = Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list").addQueryParameter("rows", 10 + "").build().getObjectObservable(FoodList.class); // 两个 Observable 的泛型应当保持一致 
            Observable.concat(cache, network).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<FoodList>() {
                @Override
                public void accept(@NonNull FoodList tngouBeen) throws Exception {
                    Log.e(TAG, "subscribe 成功:" + Thread.currentThread().getName());
                    if (isFromNet) {
                        mRxOperatorsText.append("accept : 网络获取数据设置缓存: \n");
                        Log.e(TAG, "accept : 网络获取数据设置缓存: \n" + tngouBeen.toString());
                        CacheManager.getInstance().setFoodListData(tngouBeen);
                    }
                    mRxOperatorsText.append("accept: 读取数据成功:" + tngouBeen.toString() + "\n");
                    Log.e(TAG, "accept: 读取数据成功:" + tngouBeen.toString());
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(@NonNull Throwable throwable) throws Exception {
                    Log.e(TAG, "subscribe 失败:" + Thread.currentThread().getName());
                    Log.e(TAG, "accept: 读取数据失败：" + throwable.getMessage());
                    mRxOperatorsText.append("accept: 读取数据失败：" + throwable.getMessage() + "\n");
                }
            });

## 3.3 **FlatMap** ##

![70 1][]

FlatMap的Flat是铺平的意思。即先将所有圆转成棱形，比如例子中，是先将3个圆转成6个棱形。然后，再一 一开始对它们进行转换。

public static void actionFlatMap() {
            List<Integer> list = Arrays.asList(1, 2, 3); // 这里用了lambda。 
            Observable.fromIterable(list).flatMap(integer -> {
                log("开始执行，第" + integer + "圆球的任务" + getThreadName());
                return getObservable(integer);
            }).subscribe(s -> log("已完成" + s + getThreadName()));
        }
    
        public static Observable<String> getObservable(final int integer) {
            return Observable.create((ObservableOnSubscribe<String>) emitter -> {
                emitter.onNext("第" + integer + "圆球的第1个棱形任务");
                if (integer != 1) { // 第2和第3个圆球的第二个任务延时。 
                    Thread.sleep(5 * 1000);
                }
                emitter.onNext("第" + integer + "圆球的第2个棱形任务");
                emitter.onComplete();
            }).subscribeOn(Schedulers.newThread());
        } 
    
        // 返回当前的线程名 
        public static String getThreadName() {
            return "  |  ThreadName=" + Thread.currentThread().getName();
        }
    
        private static void log(String log) {
            Log.d("FlatMap", log);
        }

运行结果：

D/TransformingOperations: 开始执行，第1圆球的任务  |  ThreadName=main 
    D/TransformingOperations: 开始执行，第2圆球的任务  |  ThreadName=main 
    D/TransformingOperations: 开始执行，第3圆球的任务  |  ThreadName=main 
    D/TransformingOperations: 已完成第2圆球的第1个棱形任务  |  ThreadName=RxNewThreadScheduler-2 
    D/TransformingOperations: 已完成第3圆球的第1个棱形任务  |  ThreadName=RxNewThreadScheduler-3 
    D/TransformingOperations: 已完成第1圆球的第1个棱形任务  |  ThreadName=RxNewThreadScheduler-3 
    D/TransformingOperations: 已完成第1圆球的第2个棱形任务  |  ThreadName=RxNewThreadScheduler-3 
    D/TransformingOperations: 已完成第2圆球的第2个棱形任务  |  ThreadName=RxNewThreadScheduler-2 
    D/TransformingOperations: 已完成第3圆球的第2个棱形任务  |  ThreadName=RxNewThreadScheduler-2

从打印的结果可以看到，FlatMap，首先是从1到3，即从左到右，执行任务的。其中，1、2、3又各自包含2个子任务。虽然是从1到3开始执行。但是，很明显，未必就是1先执行完毕。反而是，2、3的第一个任务先完成，然后，才是1的两个任务完成。然后，才是2、3的两个被延时处理的任务被分别完成。从这个例子，我们得出这样一个结论：FlatMap执行流程是：先将所有一级任务，铺平成所有二级任务。再依照，从左到右到执行次序，执行任务。但是，任务成功的回调，却不是从左到右的。而是，谁先完成谁先回调。简言之，即：**执行次序是一定的，完成次序是不确定的**。

## 3.4 concatMap ##

`concatMap` 与 `FlatMap` 的唯一区别就是 `concatMap` 保证了顺序。

## 3.5 zip ##

`zip` 操作符可以将多个 `Observable` 的数据结合为一个数据源再发射出去。

可实现多个接口数据共同更新 UI。

Observable<MobileAddress> observable1 = Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512").build().getObjectObservable(MobileAddress.class);
            Observable<CategoryResult> observable2 = Network.getGankApi().getCategoryData("Android", 1, 1);
            Observable.zip(observable1, observable2, new BiFunction<MobileAddress, CategoryResult, String>() {
                @Override
                public String apply(@NonNull MobileAddress mobileAddress, @NonNull CategoryResult categoryResult) throws Exception {
                    return "合并后的数据为：手机归属地：" + mobileAddress.getResult().getMobilearea() + "人名：" + categoryResult.results.get(0).who;
                }
            }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
                @Override
                public void accept(@NonNull String s) throws Exception {
                    Log.e(TAG, "accept: 成功：" + s + "\n");
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(@NonNull Throwable throwable) throws Exception {
                    Log.e(TAG, "accept: 失败：" + throwable + "\n");
                }
            });

适用场景：当多个请求同时满足时，进行下一步。

## 3.6 interval ##

interval 操作符实现心跳间隔任务。

private Disposable mDisposable;
    
        protected void doSomething() {
            mDisposable = Flowable.interval(1, TimeUnit.SECONDS).doOnNext(new Consumer<Long>() {
                @Override
                public void accept(@NonNull Long aLong) throws Exception {
                    Log.e(TAG, "accept: doOnNext : " + aLong);
                }
            }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Long>() {
                @Override
                public void accept(@NonNull Long aLong) throws Exception {
                    Log.e(TAG, "accept: 设置文本 ：" + aLong);
                    mRxOperatorsText.append("accept: 设置文本 ：" + aLong + "\n");
                }
            });
        }
    
        /**
         * 销毁时停止心跳
         */
        @Override
        protected void onDestroy() {
            super.onDestroy();
            if (mDisposable != null) {
                mDisposable.dispose();
            }
        }

但需要注意的是，`timer` 和 `interval` 均默认在新线程。

## 3.7 distinct ##

![620][]

这个操作符非常的简单、通俗、易懂，就是简单的去重。

Observable.just(1, 1, 1, 2, 2, 3, 4, 5).distinct().subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    mRxOperatorsText.append("distinct : " + integer + "\n");
                    Log.e(TAG, "distinct : " + integer + "\n");
                }
            });

## 3.8 Filter ##

![620 1][]

它的作用也很简单，过滤器嘛。可以接受一个参数，让其过滤掉不符合我们条件的值。

Observable.just(1, 20, 65, -5, 7, 19).filter(new Predicate<Integer>() {
                @Override
                public boolean test(@NonNull Integer integer) throws Exception {
                    return integer >= 10;
                }
            }).subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    mRxOperatorsText.append("filter : " + integer + "\n");
                    Log.e(TAG, "filter : " + integer + "\n");
                }
            });

## 3.9 buffer ##

`buffer` 操作符接受两个参数，`buffer(count,skip)`，作用是将 `Observable` 中的数据按 `skip` (步长) 分成最大不超过 count 的 `buffer` ，然后生成一个 `Observable` 。

Observable.just(1, 2, 3, 4, 5).buffer(3, 2).subscribe(new Consumer<List<Integer>>() {
                @Override
                public void accept(@NonNull List<Integer> integers) throws Exception {
                    mRxOperatorsText.append("buffer size : " + integers.size() + "\n");
                    Log.e(TAG, "buffer size : " + integers.size() + "\n");
                    mRxOperatorsText.append("buffer value : ");
                    Log.e(TAG, "buffer value : ");
                    for (Integer i : integers) {
                        mRxOperatorsText.append(i + "");
                        Log.e(TAG, i + "");
                    }
                    mRxOperatorsText.append("\n");
                    Log.e(TAG, "\n");
                }
            });

输出：

![759][]

如图，我们把 1, 2, 3, 4, 5 依次发射出来，经过 `buffer` 操作符，其中参数 `skip` 为 2， `count` 为 3，而我们的输出 依次是 123，345，5。显而易见，我们 `buffer` 的第一个参数是 `count`，代表最大取值，在事件足够的时候，一般都是取 `count` 个值，然后每次跳过 `skip` 个事件。

## 3.10 timer ##

![625][]

延时任务。

但需要注意的是，`timer` 和 `interval` 均默认在新线程。

## 3.11 doOnNext ##

它的作用是让订阅者在接收到数据之前干点有意思的事情。假如我们在获取到数据之前想先保存一下它，无疑我们可以这样实现。

Observable.just(1, 2, 3, 4).doOnNext(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    Log.e(TAG, "doOnNext 保存 " + integer + "成功" + "\n");
                }
            }).subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    Log.e(TAG, "doOnNext :" + integer + "\n");
                }
            });

输出：

![839][]

## 3.12 skip ##

`skip` 很有意思，其实作用就和字面意思一样，接受一个 long 型参数 count ，代表跳过 count 个数目开始接收。

![621][]

Observable.just(1, 2, 3, 4, 5).skip(2).subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    Log.e(TAG, "skip : " + integer + "\n");
                }
            });

输出：

![729][]

## 3.13 take ##

`take`，接受一个 long 型参数 count ，代表至多接收 count 个数据。

![621 1][]

Flowable.fromArray(1, 2, 3, 4, 5)
                    .take(2)
                    .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    Log.e(TAG, "accept: take : " + integer + "\n");
                }
            });

输出：

![797][]

## 3.14 just ##

`just`，没什么好说的，其实在前面各种例子都说明了，就是一个简单的发射器依次调用 `onNext()` 方法。

![622][]

Observable.just("1", "2", "3")
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<String>() {
                @Override
                public void accept(@NonNull String s) throws Exception {
                    Log.e(TAG, "accept : onNext : " + s + "\n");
                }
            });

输出：

![812][]

## 3.15 Single ##

顾名思义，`Single` 只会接收一个参数，而 `SingleObserver` 只会调用 `onError()` 或者 `onSuccess()`。

Single.just(new Random().nextInt()).subscribe(new SingleObserver<Integer>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                }
    
                @Override
                public void onSuccess(@NonNull Integer integer) {
                    Log.e(TAG, "single : onSuccess : " + integer + "\n");
                }
    
                @Override
                public void onError(@NonNull Throwable e) {
                    Log.e(TAG, "single : onError : " + e.getMessage() + "\n");
                }
            });

输出：

![890][]

## 3.16 debounce ##

去除发送频率过快的项，看起来好像没啥用处，但你信我，后面绝对有地方很有用武之地。

![618][]

Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception { // send events with simulated time wait
                    emitter.onNext(1); // skip
                    Thread.sleep(400);
                    emitter.onNext(2); // deliver
                    Thread.sleep(505);
                    emitter.onNext(3); // skip
                    Thread.sleep(100);
                    emitter.onNext(4); // deliver
                    Thread.sleep(605);
                    emitter.onNext(5); // deliver
                    Thread.sleep(510);
                    emitter.onComplete();
                }
            }).debounce(500, TimeUnit.MILLISECONDS)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    Log.e(TAG, "debounce :" + integer + "\n");
                }
            });

输出：

![819][]

代码很清晰，去除发送间隔时间小于 500 毫秒的发射事件，所以 1 和 3 被去掉了。

## 3.17 defer ##

简单地时候就是每次订阅都会创建一个新的 `Observable`，并且如果没有被订阅，就不会产生新的 `Observable`。

![611][]

Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<Integer>>() {
                @Override
                public ObservableSource<Integer> call() throws Exception {
                    return Observable.just(1, 2, 3);
                }
            });
            observable.subscribe(new Observer<Integer>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                }
    
                @Override
                public void onNext(@NonNull Integer integer) {
                    Log.e(TAG, "defer : " + integer + "\n");
                }
    
                @Override
                public void onError(@NonNull Throwable e) {
                    Log.e(TAG, "defer : onError : " + e.getMessage() + "\n");
                }
    
                @Override
                public void onComplete() {
                    Log.e(TAG, "defer : onComplete\n");
                }
            });

输出：

![793][]

## 3.18 last ##

`last` 操作符仅取出可观察到的最后一个值，或者是满足某些条件的最后一项。

![630][]

Observable.just(1, 2, 3)
                    .last(1)
                    .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    Log.e(TAG, "last : " + integer + "\n");
                }
            });

输出：

![736][]

唯一参数是一个默认值。

## 3.19 merge ##

`merge` 顾名思义，熟悉版本控制工具的你一定不会不知道 merge 命令，而在 Rx 操作符中，`merge` 的作用是把多个 `Observable` 结合起来，接受可变参数，也支持迭代器集合。注意它和 `concat` 的区别在于，不用等到 发射器 A 发送完所有的事件再进行发射器 B 的发送。

![620 2][]

## 3.20 reduce ##

`reduce` 操作符每次用一个方法处理一个值，可以有一个 `seed` 作为初始值

![617][]

Observable.just(1, 2, 3).reduce(new BiFunction<Integer, Integer, Integer>() {
                @Override
                public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
                    return integer + integer2;
                }
            }).subscribe(new Consumer<Integer>() {
                @Override
                public void accept(@NonNull Integer integer) throws Exception {
                    Log.e(TAG, "accept: reduce : " + integer + "\n");
                }
            });

输出：

![827][]

可以看到，代码中，我们中间采用 reduce ，支持一个 function 为两数值相加，所以应该最后的值是：1 + 2 = 3 + 3 = 6 ， 而Log 日志完美解决了我们的问题。每次两个元素处理，最后输出一个结果。

## 3.21 scan ##

`scan` 操作符作用和上面的 `reduce` 一致，唯一区别是 `reduce` 是个只追求结果的坏人，而 `scan` 会始终如一地把每一个步骤都输出。

![620 3][]

Observable.just(1, 2, 3)
                    .scan(new BiFunction<Integer, Integer, Integer>() {
                        @Override
                        public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
                            return integer + integer2;
                        }
                    })
                    .subscribe(new Consumer<Integer>() {
                        @Override
                        public void accept(@NonNull Integer integer) throws Exception {
                            Log.e(TAG, "accept: scan " + integer + "\n");
                        }
                    });

输出：

![773][]

## 3.22 window ##

按照实际划分窗口，将数据发送给不同的 `Observable。`

Observable.interval(1, TimeUnit.SECONDS) // 间隔一秒发一次
                    .take(15) // 最多接收15个
                    .window(3, TimeUnit.SECONDS)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<Observable<Long>>() {
                        @Override
                        public void accept(@NonNull Observable<Long> longObservable) throws Exception {
                            Log.e(TAG, "Sub Divide begin...\n");
    
                            longObservable.subscribeOn(Schedulers.io())
                                    .observeOn(AndroidSchedulers.mainThread())
                                    .subscribe(new Consumer<Long>() {
                                        @Override
                                        public void accept(@NonNull Long aLong) throws Exception {
                                            Log.e(TAG, "Next:" + aLong + "\n");
                                        }
                                    });
                        }
                    });

输出：

![811][]

## 3.23 takeUtil ##

![takeUntil.png][]

TakeUntil使用一个标志Observable是否发射数据来判断，当标志Observable没有发射数据时，正常发射数据，而一旦标志Observable发射过了数据则后面的数据都会被丢弃。

Observable.interval(1, TimeUnit.SECONDS)
                    .takeUntil(Observable.timer(5, TimeUnit.SECONDS)) //延迟5s
                    .subscribe(new Consumer<Long>() {
                        @Override
                        public void accept(Long aLong) throws Exception {
                            Log.e(TAG, "accept  " + aLong);
                        }
                    });

输出：

10-18 10:50:15.052 10853-10872/com.hy.rxjava E/TAG: accept  0
    10-18 10:50:16.052 10853-10872/? E/TAG: accept  1
    10-18 10:50:17.052 10853-10872/? E/TAG: accept  2
    10-18 10:50:18.052 10853-10872/? E/TAG: accept  3

## 3.24 takeWhile ##

![693176-20161120161141123-130674910.png][]

observable.takUtil(condition)，当condition == false的时候终止，不包含临界条件的item。

Observable.just(1,2, 3).takeWhile(new Predicate<Integer>() {
                @Override
                public boolean test(Integer integer) throws Exception {
                    if (integer == 2)
                        return false;
                    return true;
                }
            }).subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.e(TAG, "accept  " + integer);
                }
            });

输出：

10-18 10:46:23.166 10262-10262/com.hy.rxjava E/TAG: accept  1

## 3.25  **skipUntil** ##

SkipUntil 订阅原始的Observable，但是忽略它的发射物，直到第二个Observable发射了一项数据那一刻，它开始发射原始Observable。

Observable.interval(1, TimeUnit.SECONDS)
                    .skipUntil(Observable.empty().delay(5, TimeUnit.SECONDS))
                    .subscribe(new Consumer<Long>() {
                        @Override
                        public void accept(@NonNull Long aLong) throws Exception {
                            Log.e(TAG, "accept: " + aLong);
                        }
                    });

输出：

10-18 15:58:38.422 18296-18316/? E/TAG: accept: 4
    10-18 15:58:39.422 18296-18316/? E/TAG: accept: 5
    10-18 15:58:40.422 18296-18316/? E/TAG: accept: 6
    10-18 15:58:41.422 18296-18316/? E/TAG: accept: 7
    10-18 15:58:42.422 18296-18316/? E/TAG: accept: 8
    ........

当 SkipUntil 中的延时5秒的Observable发射之后，才开始接收Observable原本订阅的定时Observable。

# 四、Subject #

如何理解`Subject`呢？

很容易，在RxJava里面，`Observable`是数据的发射者，它会对外发射数据，然后经过`map`、`flatmap`等等数据处理后，最终传递给`Observer`，这个数据接收者。因此，抛开中间数据处理不管，可以看出，`Observable`对外发射数据，是数据流的开端；`Observer`接收数据，是数据流的末端。

那么`Subject`呢？看一眼源码：

/**
     * Represents an object that is both an Observable and an Observer.
     */
    public abstract class Subject<T, R> extends Observable<R> implements Observer<T> {}

首先，它`extends Observable<R>`，说明`Subject`具备了对外发射数据的能力，即拥有了`from()`、`just()`等等；另外，它又`implements Observer<T>`，说明又能够处理数据，具备`onNext()`、`onCompleted`等等。

然后，`Subject`毕竟只是一个抽象类，那么我们要如何使用它呢？

这里介绍一种最简单的：`PublishSubject`:

PublishSubject<Object> subject = PublishSubject.create();
      // myObserver will receive "one" & "two" and onCompleted events
      subject.subscribe(myObserver);
      subject.onNext("one");
      subject.onNext("two");
      subject.onCompleted();

这里做的事情很简单，先创建一个`PublishSubject` \-> 绑定一个`myObserver`，此时`subject`扮演了`Observable`的角色，把数据发射给`myObserver` \-> 然后`subject`处理接收了两个数据`one`、`two` \-> 最终这些数据都传递给了`myObserver`。所以，`subject`扮演的角色是:

**数据`one`、`two` => (Observer) `subject` (Observable) => `myObserver`**

简单来说，我们把数据`one`、`two`塞给`subject`，然后`subject`又发射给了`myObserver`。

# 五、Transformer与compose #

Transformer，顾名思义是转换器的意思。

早在 RxJava1.x 版本就有了Observable.Transformer、Single.Transformer和Completable.Transformer，在2.x版本中变成了ObservableTransformer、SingleTransformer、CompletableTransformer、FlowableTransformer和MaybeTransformer。其中，FlowableTransformer和MaybeTransformer是新增的。由于 RxJava2 将Observable拆分成 Observable 和 Flowable，所以多了一个FlowableTransformer。同时，Maybe是 RxJava2 新增的一个类型，所以多了MaybeTransformer。为啥要说 RxJava1.x 版本呢，因为我发现网上很多代码都是 RxJava1.x 版本的写法，对不了解的同学造成了很多误解。

Transformer 能够将一个 Observable/Flowable/Single/Completable/Maybe 对象转换成另一个 Observable/Flowable/Single/Completable/Maybe 对象，和调用一系列的内联操作符是一模一样的。

举个简单的例子，写一个transformer()方法将一个发射整数的Observable转换为发射字符串的Observable。

public static ObservableTransformer<Integer, String> transformer() {
            return new ObservableTransformer<Integer, String>() {
                @Override
                public ObservableSource<String> apply(@NonNull Observable<Integer> upstream) {
                    return upstream.map(new Function<Integer, String>() {
                        @Override
                        public String apply(@NonNull Integer integer) throws Exception {
                            return integer+"a";
                        }
                    });
                }
            };
        }

调用：

Observable.just(123, 456)
                    .compose(transformer())
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(@io.reactivex.annotations.NonNull String s) throws Exception {
                            Log.e(TAG, "s=" + s);
                        }
                    });

输出：

10-18 14:26:33.846 14536-14536/com.hy.rxjava E/TAG: s=123a
    10-18 14:26:33.846 14536-14536/com.hy.rxjava E/TAG: s=456a

通常用Transformer与compose配合，精简链式调用，比如线程切换：

public static <T> ObservableTransformer <T, T> lifeTransFormer(){
            ObservableTransformer<T, T> observableTransformer = new ObservableTransformer<T, T>() {
    
                @Override
                public ObservableSource<T> apply(Observable<T> upstream) {
                    return  upstream.subscribeOn(Schedulers.io())
                            .observeOn(AndroidSchedulers.mainThread());
                }
            };
            return observableTransformer;
        }

调用：

Observable.just(1)
                    .doOnNext(new Consumer<Integer>() {
                        @Override
                        public void accept(Integer integer) throws Exception {
                            Log.e(TAG, "accept thread " + Thread.currentThread().getName());
                        }
                    })
                    .compose(MainActivity.<Integer>lifeTransFormer())
                    .subscribe(new Consumer<Integer>() {
                        @Override
                        public void accept(Integer integer) throws Exception {
                            Log.e(TAG, "accept thread " + Thread.currentThread().getName());
                        }
                    });

输出：

10-18 15:11:16.157 16318-16335/? E/TAG: accept thread RxCachedThreadScheduler-1
    10-18 15:11:16.174 16318-16318/? E/TAG: accept thread main

参考：

[这可能是最好的RxJava 2.x 教程（完结版）][RxJava 2.x]

[Android Rxjava：这是一篇 清晰 & 易懂的Rxjava 入门教程 ][Android Rxjava_ _ _ _Rxjava _]

[RxJava 的 CompositeSubscription与内存泄漏 ][RxJava _ CompositeSubscription_]

[RxJava2 源码分析(一)][RxJava2]

[70]: /images/20220601/2352435b72184326a4955d1dc262cfd2.png
[610]: /images/20220601/bee414f68fa64868b27801a2b5a18eb4.png
[70 1]: /images/20220601/2e420140325641f2b8b13089748e34cd.png
[620]: /images/20220601/93cbecfcb7dd439785b53fbbfa51afb1.png
[620 1]: /images/20220601/958ce98198fe4592ad805c1ff6684e6d.png
[759]: /images/20220601/99a3413774fd4c6aa8e5663c247e2971.png
[625]: /images/20220601/2f08029122f24bfe9cfb7d2f1c993a6a.png
[839]: /images/20220601/ee341c3665364b529a9075039e6379b9.png
[621]: /images/20220601/094b594b701f4c8894877efd66f144f4.png
[729]: /images/20220601/40747be9f67d4c608a2711161f3762e3.png
[621 1]: /images/20220601/f54388265a1447768cb7f85ce18ee17d.png
[797]: /images/20220601/848df27191ca4a1387a672d928f91a85.png
[622]: /images/20220601/293cbbddc7ad4a4891bb98f699e4a3f0.png
[812]: /images/20220601/e073b08ff5e247559212547f5895fe0d.png
[890]: /images/20220601/45087cbe471d43318e56d90b6bffaaa6.png
[618]: /images/20220601/a118b5fa4222454b89fa4800271c73af.png
[819]: /images/20220601/02661d32bf9b40fca53469df1f9d5ddf.png
[611]: /images/20220601/75f869ece8964853bdc1fdcfe8157bdc.png
[793]: /images/20220601/40fd15705b57454fa3a1ed58bca6667f.png
[630]: /images/20220601/12defa9a3adf43049eeb582bb480c974.png
[736]: /images/20220601/de455af13bd54e7486a4ae5b0a11da04.png
[620 2]: /images/20220601/1e461181fa214f44b3c584b1d31b43b3.png
[617]: /images/20220601/e80347209f064a92b6af095f4d8b1980.png
[827]: /images/20220601/fd8236b7293a427da42961db3d949d1a.png
[620 3]: /images/20220601/dc464a5011cf429fb088d286569c872c.png
[773]: /images/20220601/ae1497b41ab84e8fab51a4e5bf37868f.png
[811]: /images/20220601/85355d70d89c47c591e4e61b6fe6d728.png
[takeUntil.png]: https://raw.githubusercontent.com/wiki/ReactiveX/RxJava/images/rx-operators/takeUntil.png
[693176-20161120161141123-130674910.png]: /images/20220601/abf4d13c9f394dba96166c8c0ffbdfa2.png
[RxJava 2.x]: https://www.jianshu.com/p/0cd258eecf60
[Android Rxjava_ _ _ _Rxjava _]: https://www.jianshu.com/p/a406b94f3188
[RxJava _ CompositeSubscription_]: https://blog.csdn.net/zr940326/article/details/51550002
[RxJava2]: https://www.jianshu.com/p/a488318dec0e