这可能是最好的 RxJava 2.x 入门教程系列专栏
文章链接：
这可能是最好的 RxJava 2.x 入门教程(完结版)【重磅推出】
这可能是最好的 RxJava 2.x 入门教程（一）
这可能是最好的 RxJava 2.x 入门教程（二）
这可能是最好的 RxJava 2.x 入门教程（三）
这可能是最好的 RxJava 2.x 入门教程（四）
这可能是最好的 RxJava 2.x 入门教程（五）
GitHub 代码同步更新：https://github.com/nanchen2251/RxJava2Examples
为了满足大家的饥渴难耐，GitHub 将同步更新代码，主要包含基本的代码封装，RxJava 2.x 所有操作符应用场景介绍和实际应用场景，后期除了 RxJava 可能还会增添其他东西，总之，GitHub 上的 Demo 专为大家倾心打造。传送门：https://github.com/nanchen2251/RxJava2Examples
前言
年轻的老司机们，我这么勤的为大家分享，却少有催更的，好吧。其实写这个系列不是为了吸睛，那咱们继续写我们的 RxJava 2.x 的操作符。
正题
distinct
这个操作符非常的简单、通俗、易懂，就是简单的去重嘛，我甚至都不想贴代码，但人嘛，总得持之以恒。
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");
}
});
输出：
Log 日志显而易见，我们在经过
dinstinct() 后接收器接收到的事件只有1,2,3,4,5了。
Filter
信我，Filter 你会很常用的，它的作用也很简单，过滤器嘛。可以接受一个参数，让其过滤掉不符合我们条件的值
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");
}
});
输出：
可以看到，我们过滤器舍去了小于 10 的值，所以最好的输出只有 20, 65, 19。
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");
}
});
输出：
如图，我们把 1, 2, 3, 4, 5 依次发射出来，经过
buffer 操作符，其中参数 skip 为 2， count 为 3，而我们的输出 依次是 123，345，5。显而易见，我们 buffer 的第一个参数是 count，代表最大取值，在事件足够的时候，一般都是取 count 个值，然后每次跳过 skip 个事件。其实看 Log 日志，我相信大家都明白了。
timer
timer 很有意思，相当于一个定时任务。在 1.x 中它还可以执行间隔逻辑，但在 2.x 中此功能被交给了 interval，下一个会介绍。但需要注意的是，timer 和 interval 均默认在新线程。
mRxOperatorsText.append("timer start : " + TimeUtil.getNowStrTime() + "\n");
Log.e(TAG, "timer start : " + TimeUtil.getNowStrTime() + "\n");
Observable.timer(2, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()) // timer 默认在新线程，所以需要切换回主线程
.subscribe(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
mRxOperatorsText.append("timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
Log.e(TAG, "timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
}
});
输出：
显而易见，当我们两次点击按钮触发这个事件的时候，接收被延迟了 2 秒。
interval
如同我们上面可说，interval 操作符用于间隔时间执行某个操作，其接受三个参数，分别是第一次发送延迟，间隔时间，时间单位。
mRxOperatorsText.append("interval start : " + TimeUtil.getNowStrTime() + "\n");
Log.e(TAG, "interval start : " + TimeUtil.getNowStrTime() + "\n");
Observable.interval(3,2, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()) // 由于interval默认在新线程，所以我们应该切回主线程
.subscribe(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
mRxOperatorsText.append("interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
Log.e(TAG, "interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
}
});
输出：
如同 Log 日志一样，第一次延迟了 3 秒后接收到，后面每次间隔了 2 秒。
然而，心细的小伙伴可能会发现，由于我们这个是间隔执行，所以当我们的Activity 都销毁的时候，实际上这个操作还依然在进行，所以，我们得花点小心思让我们在不需要它的时候干掉它。查看源码发现，我们subscribe(Cousumer<? super T> onNext)返回的是Disposable，我们可以在这上面做文章。
@Override
protected void doSomething() {
mRxOperatorsText.append("interval start : " + TimeUtil.getNowStrTime() + "\n");
Log.e(TAG, "interval start : " + TimeUtil.getNowStrTime() + "\n");
mDisposable = Observable.interval(3, 2, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()) // 由于interval默认在新线程，所以我们应该切回主线程
.subscribe(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
mRxOperatorsText.append("interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
Log.e(TAG, "interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
}
});
}
@Override
protected void onDestroy() {
super.onDestroy();
if (mDisposable != null && !mDisposable.isDisposed()) {
mDisposable.dispose();
}
}
哈哈，再次验证，解决了我们的疑惑。
doOnNext
其实觉得 doOnNext 应该不算一个操作符，但考虑到其常用性，我们还是咬咬牙将它放在了这里。它的作用是让订阅者在接收到数据之前干点有意思的事情。假如我们在获取到数据之前想先保存一下它，无疑我们可以这样实现。
Observable.just(1, 2, 3, 4)
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
mRxOperatorsText.append("doOnNext 保存 " + integer + "成功" + "\n");
Log.e(TAG, "doOnNext 保存 " + integer + "成功" + "\n");
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
mRxOperatorsText.append("doOnNext :" + integer + "\n");
Log.e(TAG, "doOnNext :" + integer + "\n");
}
});
输出：
skip
skip 很有意思，其实作用就和字面意思一样，接受一个 long 型参数 count ，代表跳过 count 个数目开始接收。
Observable.just(1,2,3,4,5)
.skip(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
mRxOperatorsText.append("skip : "+integer + "\n");
Log.e(TAG, "skip : "+integer + "\n");
}
});
输出：
take
take，接受一个 long 型参数 count ，代表至多接收 count 个数据。
Flowable.fromArray(1,2,3,4,5)
.take(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
mRxOperatorsText.append("take : "+integer + "\n");
Log.e(TAG, "accept: take : "+integer + "\n" );
}
});
输出：
just
just，没什么好说的，其实在前面各种例子都说明了，就是一个简单的发射器依次调用 onNext() 方法。
Observable.just("1", "2", "3")
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
mRxOperatorsText.append("accept : onNext : " + s + "\n");
Log.e(TAG,"accept : onNext : " + s + "\n" );
}
});
输出：
写在最后
好吧，本节先讲到这里，下节我们还是继续讲简单的操作符，虽然我们的教程比较枯燥，现在也不那么受人关注，但后面的系列我相信大家一定会非常喜欢的，我们下期再见！
代码全部同步到GitHub：https://github.com/nanchen2251/RxJava2Examples
做不完的开源，写不完的矫情。欢迎扫描下方二维码或者公众号搜索「nanchen」关注我的微信公众号，目前多运营 Android ，尽自己所能为你提升。如果你喜欢，为我点赞分享吧~
nanchen
网友评论
1. 关于interval和timer的讲解。。语言组织的不太好啊。。我看interval的时候一直在疑惑为啥没有Observable的创建过程以及数据的发送过程。。直到点到官网链接才看懂。。
create an Observable that emits a sequence of integers spaced by a given time interval。你应该把官网这句讲解加上的。
2. mRxOperatorsText这是一个TextView，用于展示你的代码运行结果的。也应该在文章开头加个解释的。。我是下载了源码才看懂，最开始以为是个String字符串。。
count 是每组一共有多少个发射, skip是每组之间发射的间隔
如: 1,2,3,4,5 buffer(3,2)
第一组 就是 1,2,3
第二组 就是 第一组基础上跳过2个发射的步长 也就是从3 开始 3,4,5
第三组 在第二组基础上 跳过2个发射步长 也就是从5开始 5
若buffer(4,1)
第一组 步长从零 1 2 3 4
第二组 步长+1 2 3 4 5
第三组 步长+1 3 4 5
第四组 步长+1 4 5
第五组 步长+1 5