Rxjava2 操作符 - Filtering Observab
2017-10-25 本文已影响74人
joker_fu
本教程均是基于java的项目：
Debounce — 在固定的时间没有新的项目产生，就把这个项目从Observable 发送出去。Debounce操作符会过滤掉发射速率过快的数据项。
private static void debounce() {
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
int time;
for (int i = 1; i < 11; i++) {
Thread.sleep(time = 100 * i);
e.onNext("emit on " + time);
}
e.onComplete();
}
})
.subscribeOn(Schedulers.io())
.debounce(900, TimeUnit.MILLISECONDS)
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(@NonNull String s) {
System.out.println("onNext: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
System.out.println("onError");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
while (true) ;
}
输出结果：
onSubscribe
onNext: emit on 900
onNext: emit on 1000
onComplete
Distinct — 抑制Observable产生的重复项目。Distinct的过滤规则是：只允许还没有发射过的数据项通过。
private static void distinct() {
Observable
.just(0, 1, 1, 3, 4, 5, 4, 6)
.distinct()
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("onNext: " + integer);
}
});
}
输出结果：
onNext: 0
onNext: 1
onNext: 3
onNext: 4
onNext: 5
onNext: 6
ElementAt — Observable只发送第N个项目。ElementAt操作符获取原始Observable发射的数据序列指定索引位置的数据项，然后当做自己的唯一数据发射。
private static void elementAt() {
Observable
.just(0, 1, 1, 3, 4, 5, 4, 6)
.elementAt(3)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println(integer);
}
});
}
输出结果：
3
Filter — Observable 只发送通过Predicate测试的项目。
private static void filter() {
Observable
.just(0, 1, 2, 3, 4, 5)
.filter(new Predicate<Integer>() {
@Override
public boolean test(@NonNull Integer integer) throws Exception {
return integer % 2 == 1;
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 1
next: 3
next: 5
First — Observable只发送第一个项目或第一个满足条件的项目。
private static void first() {
Observable
.just(0, 1, 2, 3, 4, 5)
.first(-1)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 0
IgnoreElements — 不发送任何项目，只关注结果
private static void ignoreElements() {
Observable
.just(0, 1, 2, 3, 4, 5)
.ignoreElements()
.subscribe(new CompletableObserver() {
@Override
public void onSubscribe(@NonNull Disposable d) {
System.out.println("onSubscribe ");
}
@Override
public void onComplete() {
System.out.println("onComplete ");
}
@Override
public void onError(@NonNull Throwable e) {
System.out.println("onError ");
}
});
}
输出结果：
onSubscribe
onComplete
Last — 只通过 Observable发送最后一个项目
private static void last() {
Observable
.just(0, 1, 2, 3, 4, 5)
.last(-10)
.subscribe(new BiConsumer<Integer, Throwable>() {
@Override
public void accept(Integer integer, Throwable throwable) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 5
Sample — 在规定的时间，发送最新的项目
private static void sample() {
Observable
.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; i < 10; i++) {
e.onNext(i);
Thread.sleep(1000);
}
}
})
.sample(2, TimeUnit.SECONDS)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 1
next: 3
next: 5
next: 7
next: 9
Skip —Observable 跳过先前N个项目发送
private static void skip() {
Observable
.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; i < 10; i++) {
e.onNext(i);
Thread.sleep(1000);
}
}
})
.skip(5)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 5
next: 6
next: 7
next: 8
next: 9
SkipLast — Observable 跳过接下来的N个项目发送
private static void skip() {
Observable
.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; i < 10; i++) {
e.onNext(i);
Thread.sleep(1000);
}
}
})
.skipLast(5,TimeUnit.SECONDS)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 0
next: 1
next: 2
next: 3
next: 4
Take — Observable 仅仅发送先前的N个项目
private static void take() {
Observable
.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 0)
.take(5)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 1
next: 2
next: 3
next: 4
next: 5
TakeLast —Observable 仅仅发送接下来的N个项目
private static void take() {
Observable
.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 0)
.takeLast(5)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("next: " + integer);
}
});
}
输出结果：
next: 6
next: 7
next: 8
next: 9
next: 0
