RxJava中还有许多操作符。操作符就是用于在Observable和最终的Observer之间，通过转换Observable为其他观察者对象的过程，修改发出的事件，最终将最简洁的数据传递给Observer对象。下面我们介绍一些比较常用的操作符。
-
just()方式
Observable<String> observable = Observable.just("Hello");
使用just( )，将为你创建一个Observable并自动为你调用onNext( )发射数据。通过just( )方式 直接触发onNext()，just中传递的参数将直接在Observer的onNext()方法中接收到。
Observable.just(1, 2, 3, 4)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "Observer log is : " + integer);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
});
打印结果如下,可以看到将just括号里面的数字一个个打印
D/MainActivity: Observer log is : 1
D/MainActivity: Observer log is : 2
D/MainActivity: Observer log is : 3
D/MainActivity: Observer log is : 4
-
range( )方式
Observable<Integer> observable = Observable.range(1,20);
创建一个发射特定整数序列的Observable，第一个参数为起始值，第二个为发送的个数，如果为0则不发送，负数则抛异常。上述表示发射1到20的数。即调用20次nNext()方法，依次传入1-20数字。
-
过滤操作符 Filter
Filter,顾名思义,过滤器,可以过滤掉一部分不符合要求的事件,当上游给我们发送的数据超多,而下游需要的只是一些特定的数据,如果全部接收上游发送的数据,很容易造成OOM,为了避免OOM的出现,我们则需要对上游数据进行过滤,具体操作如下:
过滤掉出大于4的数字
Observable.just(1, 2, 0, 9, 7, 3, 4)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer > 4;
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "Observer log is : " + integer);
}
});
-
Delay 延迟发射
Observable.fromArray(1, 2, 3)
.delay(3, TimeUnit.SECONDS)//3s以后再去发射事件 整体延迟 区别于interval对每个事件的延迟效果
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "Observer log is : " + integer);
}
});
-
timer( )方式
Observable<Integer> observable = Observable.timer(2, TimeUnit.SECONDS);
创建一个Observable，它在一个给定的延迟后发射一个特殊的值，即表示延迟2秒后，调用onNext()方法。
Observable.timer(5, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Long aLong) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
-
doOnNext()
doOnNext()允许我们在每次输出一个元素之前做一些额外的事情。
Observable.just(1, 2, 3, 4, 5)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "准备做其他的事情 : ");
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "最后输出的内容 : " + integer);
}
});
输出结果为 可以看到 在输出结果之前执行了doOnNext ，我们便可以在此做些其他的操作
D/MainActivity: 准备做其他的事情 :
D/MainActivity: 最后输出的内容 : 1
D/MainActivity: 准备做其他的事情 :
D/MainActivity: 最后输出的内容 : 2
D/MainActivity: 准备做其他的事情 :
D/MainActivity: 最后输出的内容 : 3
D/MainActivity: 准备做其他的事情 :
D/MainActivity: 最后输出的内容 : 4
D/MainActivity: 准备做其他的事情 :
D/MainActivity: 最后输出的内容 : 5
-
take()操作符
内部通过OperatorTake和OperatorTakeTimed过滤数据。
只发射开始的N项数据或者一定时间内的数据。
Observable.just(1, 2, 3, 4, 5)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.take(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "最后输出的内容 : " + integer);
}
});
-
takeLast()操作符
只发射最后的N项数据或者一定时间内的数据。内部通过OperatorTakeLast和OperatorTakeLastTimed过滤数据。takeLastBuffer和takeLast类似，不同点在于takeLastBuffer会收集成List后发射。
-
interval()操作符
创建一个按照给定的时间间隔发射从0开始的整数序列的Observable<Long>，内部通过OnSubscribeTimerPeriodically工作。
下面代码表示等待3秒后从0 开始轮询，每各3秒打印一次
Observable.interval(3, TimeUnit.SECONDS).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.w(TAG, "" + aLong);
}
});
-
skip()操作符
跳过开始的N项数据或者一定时间内的数据。内部通过OperatorSkip和OperatorSkipTimed实现过滤
Observable.just(1, 2, 3, 4, 5)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
. skip(2) //跳过第2项
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "最后输出的内容 : " + integer);
}
});
-
skipLast()操作符
跳过最后的N项数据或者一定时间内的数据。内部通过OperatorSkipLast和OperatorSkipLastTimed实现过滤。
-
distinct()操作符
Observable.just(1, 2, 3, 4, 3, 3, 4, 5)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.distinct()
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "最后输出的内容 : " + integer);
}
});
