1、filter First
StartWith操作符会在源Observable发射的数据前面插上一些数据。
Observable.range(1, 2).startWith(0).subscribe(consumer);
//Observable.range(1, 2).startWithArray(-1,0).subscribe(consumer);
12-29 13:27:50.680 6062-6062/com.rxjavademo E/tag: 0
12-29 13:27:50.686 6062-6062/com.rxjavademo E/tag: 1
12-29 13:27:50.689 6062-6062/com.rxjavademo E/tag: 2
First 操作符返回第一条数据或者返回满足条件的第一条数据。
Last操作符返回最后一条数据或者满足条件的最后一条数据。
2、elementAt
elementAt(index） 发送脚标为index的数据
//在这里发送0、1、2、3、4，脚标为3的数据为3
Observable.range(0, 5).elementAt(3).subscribe(consumer);
12-29 12:01:41.090 461-461/com.rxjavademo E/tag: 3
3、distinct
distinct 去重
//
Observable.just(1, 1, 2, 2, 2, 3).distinct().subscribe(consumer);
12-29 12:05:34.576 461-461/com.rxjavademo E/tag: 1
12-29 12:05:34.586 461-461/com.rxjavademo E/tag: 2
12-29 12:05:34.590 461-461/com.rxjavademo E/tag: 3
distinctUntilChanged操作符用来过滤掉连续的重复数据。
//
Observable.just(1, 2, 2, 3,1).distinctUntilChanged().subscribe(consumer);
12-29 13:20:37.298 5516-5516/com.rxjavademo E/tag: 1
12-29 13:20:37.308 5516-5516/com.rxjavademo E/tag: 2
12-29 13:20:37.311 5516-5516/com.rxjavademo E/tag: 3
12-29 13:20:37.314 5516-5516/com.rxjavademo E/tag: 1
4、skip
前N项去除掉
//
Observable.range(1, 5).skip(2).subscribe(consumer);
12-29 12:11:53.557 461-461/com.rxjavademo E/tag: 3
12-29 12:11:53.563 461-461/com.rxjavademo E/tag: 4
12-29 12:11:53.568 461-461/com.rxjavademo E/tag: 5
5、take
发送的元素只取前N项
//
Observable.range(0, 5).take(2).subscribe(consumer);
12-29 12:19:15.257 1642-1642/com.rxjavademo E/tag: 0
12-29 12:19:15.265 1642-1642/com.rxjavademo E/tag: 1
Observable.interval(1, TimeUnit.SECONDS, AndroidSchedulers.mainThread()).take(3).subscribe(consumer);
12-29 12:21:33.071 2269-2269/com.rxjavademo E/tag: 0
12-29 12:21:34.071 2269-2269/com.rxjavademo E/tag: 1
12-29 12:21:35.071 2269-2269/com.rxjavademo E/tag: 2
6、ignoreElements
ignoreElements操作符忽略所有源Observable产生的结果，只会执行onCpmpleted()或者onError()方法
Observable.range(0, 5).ignoreElements().subscribe(new Action() {
@Override
public void run() throws Exception {
Log.e("tag","onCompleted()" );
}
}, consumer);
12-29 12:39:59.976 3700-3700/com.rxjavademo E/tag: onCompleted()
7、throttleFirst
throttleFirst操作符，会定期发送这个时间段里源Observable发送的第一个数据
Observable
.create(new ObservableOnSubscribe<Object>() {
@Override
public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
for (int i = 0; i < 10; i++) {
emitter.onNext(i);
//线程休眠100毫秒
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
})
.throttleFirst(200, TimeUnit.MILLISECONDS)
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
12-29 13:09:25.066 3700-3700/com.rxjavademo E/tag: 0
12-29 13:09:25.258 3700-3700/com.rxjavademo E/tag: 2
12-29 13:09:25.459 3700-3700/com.rxjavademo E/tag: 4
12-29 13:09:25.662 3700-3700/com.rxjavademo E/tag: 6
12-29 13:09:25.862 3700-3700/com.rxjavademo E/tag: 8
6、throttleWithTimeout
如果两次数据的发射间隔小于指定时间，就会丢弃前一次的数据,直到指定时间内都没有新数据发射时才进行发射
Observable
.create(new ObservableOnSubscribe<Object>() {
@Override
public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
emitter.onNext(1);
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw Exceptions.propagate(e);
}
emitter.onNext(2);
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw Exceptions.propagate(e);
}
emitter.onNext(3);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw Exceptions.propagate(e);
}
emitter.onNext(4);
emitter.onNext(5);
emitter.onNext(6);
emitter.onComplete();
}
})
.throttleWithTimeout(800, TimeUnit.MILLISECONDS)
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
12-29 13:10:12.382 4834-4834/com.rxjavademo E/tag: 3
12-29 13:10:12.582 4834-4834/com.rxjavademo E/tag: 6
网友评论