RxJava中的常用操作符
2017-08-24 本文已影响0人
anvata
创建操作
- create
-
defer
直到有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable。
defer.c.png
Defer操作符会一直等待直到有观察者订阅它，然后它使用Observable工厂方法生成一个Observable。它对每个观察者都这样做，因此尽管每个订阅者都以为自己订阅的是同一个Observable，事实上每个订阅者获取的是它们自己的单独的数据序列。
在某些情况下，等待直到最后一分钟（就是直到订阅发生时）才生成Observable可以确保Observable包含最新的数据。 -
Empty/Never/Throw
Empty
创建一个不发射任何数据但是正常终止的Observable
Never
创建一个不发射数据也不终止的Observable
Throw
创建一个不发射数据以一个错误终止的Observable -
fromArray/fromIterable
将其它种类的对象和数据类型转换为Observable
from.png
在RxJava中，from操作符可以转换Iterable和Array等。对于Iterable和数组，产生的Observable会发射Iterable或数组
-
Interval
创建一个按固定时间间隔发射无线递增的整数序列的Observable,RxJava将这个操作符实现为interval方法。它接受一个表示时间间隔的参数和一个表示时间单位的参数。
interval(long,TimeUnit)
interval(long,TimeUnit,Scheduler)
-
just
创建一个发射指定值的Observable,Just将单个数据转换为发射那个数据的Observable。
Just类似于From，但是From会将数组或Iterable的数据取出然后逐个发射，而Just只是简单的原样发射，将数组或Iterable当做单个数据。just方法最多接受10个参数,返回一个按参数列表顺序发射这些数据的Observable -
Range
创建一个发射特定整数序列的Observable
RxJava将这个操作符实现为range函数，它接受两个参数，一个是范围的起始值，一个是范围的数据的数目。如果你将第二个参数设为0，将导致Observable不发射任何数据（如果设置为负数，会抛异常）。 -
Repeat
创建一个发射特定数据重复多次的Observable
RxJava将这个操作符实现为repeat方法。它不是创建一个Observable，而是重复发射原始Observable的数据序列，这个序列或者是无限的，或者通过repeat(n)指定重复次数
变换操作
-
map
对Observable发射的每一项数据应用一个函数，执行变换操作
map.png
Map操作符对原始Observable发射的每一项数据应用一个你选择的函数，然后返回一个发射这些结果的Observable。
RxJava将这个操作符实现为map函数。这个操作符默认不在任何特定的调度器上执行。
Observable.just(1,2,3)
.map(new Function<Integer, String>() {
@Override
public String apply(@NonNull Integer integer) throws Exception {
return "integer is" + integer;
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.i(TAG, "accept: s = " + s);
}
});
08-24 17:08:19.573 27421-27421/com.example.wty.learnrxjava I/MainActivity: accept: s = integer is 1
08-24 17:08:19.573 27421-27421/com.example.wty.learnrxjava I/MainActivity: accept: s = integer is 2
08-24 17:08:19.573 27421-27421/com.example.wty.learnrxjava I/MainActivity: accept: s = integer is 3
-
flatMap
FlatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable
mergeMap.png
FlatMap操作符使用一个指定的函数对原始Observable发射的每一项数据执行变换操作，这个函数返回一个本身也发射数据的Observable，然后FlatMap合并这些Observables发射的数据，最后将合并后的结果当做它自己的数据序列发射。
这个方法是很有用的，例如，当你有一个这样的Observable：它发射一个数据序列，这些数据本身包含Observable成员或者可以变换为Observable，因此你可以创建一个新的Observable发射这些次级Observable发射的数据的完整集合。
注意：FlatMap对这些Observables发射的数据做的是合并(merge)操作，因此它们可能是交错的。如果想要按照严格的顺序发射这些数据,使用ConcatMap操作符即可
举个例子:
Observable.fromIterable(getData())
.flatMap(new Function<NoteBook, ObservableSource<Note>>() {
@Override
public ObservableSource<Note> apply(@NonNull NoteBook noteBook) throws Exception {
return Observable.fromIterable(noteBook.getNotes());
}
})
.subscribe(new Consumer<Note>() {
@Override
public void accept(Note note) throws Exception {
Log.i(TAG, "accept: " + note);
}
});
08-24 17:14:09.512 32091-32091/? I/MainActivity: accept: Note{id='1', noteBookId='1', title='Introduction', content='$$$$$$$$$$$$$$$$$$'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='2', noteBookId='1', title='ReactiveX', content='########################'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='3', noteBookId='1', title='Observables', content='@@@@@@@@@@@@@@@@@@@@@@@@@@'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='4', noteBookId='1', title='Operators Categories', content='********************'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='5', noteBookId='1', title='RxJava文档和教程', content='********************'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='6', noteBookId='2', title='Retrofit入门教程1', content='$$$$$$$$$$$$$$$$$$'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='7', noteBookId='2', title='Retrofit入门教程2', content='########################'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='8', noteBookId='2', title='Retrofit入门教程3', content='@@@@@@@@@@@@@@@@@@@@@@@@@@'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='9', noteBookId='2', title='Retrofit入门教程4', content='********************'}
08-24 17:14:09.514 32091-32091/? I/MainActivity: accept: Note{id='10', noteBookId='2', title='Retrofit入门教程5', content='********************'}
-
Buffer
定期收集Observable的数据放进一个数据包裹，然后发射这些数据包裹，而不是一次发射一个值。
buffer.png
Buffer操作符将一个Observable变换为另一个，原来的Observable正常发射数据，变换产生的Observable发射这些数据的缓存集合。Buffer操作符在很多语言特定的实现中有很多种变体，它们在如何缓存这个问题上存在区别。
注意：如果原来的Observable发射了一个onError通知，Buffer会立即传递这个通知，而不是首先发射缓存的数据，即使在这之前缓存中包含了原始Observable发射的数据。
Observable.interval(1000,TimeUnit.MILLISECONDS)
.buffer(5)
.subscribe(new Consumer<List<Long>>() {
@Override
public void accept(List<Long> longs) throws Exception {
Log.i(TAG, "accept: longs = " + longs);
}
});
08-24 17:04:12.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [0, 1, 2, 3, 4]
08-24 17:04:17.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [5, 6, 7, 8, 9]
08-24 17:04:22.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [10, 11, 12, 13, 14]
08-24 17:04:27.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [15, 16, 17, 18, 19]
08-24 17:04:32.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [20, 21, 22, 23, 24]
08-24 17:04:37.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [25, 26, 27, 28, 29]
-
GroupBy
将一个Observable分拆为一些Observables集合，它们中的每一个发射原始Observable的一个子序列
groupBy.c.png
GroupBy操作符将原始Observable分拆为一些Observables集合，它们中的每一个发射原始Observable数据序列的一个子序列。哪个数据项由哪一个Observable发射是由一个函数判定的，这个函数给每一项指定一个Key，Key相同的数据会被同一个Observable发射。
举个例子,
Observable.interval(1, TimeUnit.SECONDS)
.take(30)
.groupBy(new Function<Long, Integer>() {
@Override
public Integer apply(@NonNull Long aLong) throws Exception {
if (aLong % 3 == 0) {
return 3;
} else if (aLong % 4 == 0) {
return 4;
} else if (aLong % 5 == 0) {
return 5;
} else {
return 1;
}
}
})
.subscribe(new Consumer<GroupedObservable<Integer, Long>>() {
@Override
public void accept(GroupedObservable<Integer, Long> longLongGroupedObservable) throws Exception {
Integer key = longLongGroupedObservable.getKey();
Log.i(TAG, "accept: key = " + key);
if (key == 3) {
longLongGroupedObservable.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.i(TAG, "accept: key = 3,3的倍数,aLong = " + aLong);
}
});
} else if (key == 4) {
longLongGroupedObservable.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.i(TAG, "accept: key = 4,4的倍数 aLong = " + aLong);
}
});
} else if (key == 5) {
longLongGroupedObservable.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.i(TAG, "accept: key = 5,5的倍数 aLong = " + aLong);
}
});
} else {
longLongGroupedObservable.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.i(TAG, "accept: key = 1,不是3、4、5的倍数 aLong = " + aLong);
}
});
}
}
});
08-24 17:22:37.134 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1
08-24 17:22:37.135 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 1
08-24 17:22:37.231 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 2
08-24 17:22:37.331 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3
08-24 17:22:37.332 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 3
08-24 17:22:37.431 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4
08-24 17:22:37.432 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 4
08-24 17:22:37.531 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5
08-24 17:22:37.532 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5,5的倍数 aLong = 5
08-24 17:22:37.631 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 6
08-24 17:22:37.731 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 7
08-24 17:22:37.831 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 8
08-24 17:22:37.931 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 9
08-24 17:22:38.031 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5,5的倍数 aLong = 10
08-24 17:22:38.131 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 11
08-24 17:22:38.231 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 12
08-24 17:22:38.331 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 13
08-24 17:22:38.431 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 14
08-24 17:22:38.531 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 15
08-24 17:22:38.631 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 16
08-24 17:22:38.731 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 17
08-24 17:22:38.831 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 18
08-24 17:22:38.931 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 19
08-24 17:22:39.031 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 20
08-24 17:22:39.131 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 21
08-24 17:22:39.231 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 22
08-24 17:22:39.331 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 23
08-24 17:22:39.431 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 24
08-24 17:22:39.531 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5,5的倍数 aLong = 25
08-24 17:22:39.631 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 26
08-24 17:22:39.731 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 27
08-24 17:22:39.831 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 28
08-24 17:22:39.932 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 29
08-24 17:22:40.031 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 30
过滤操作
-
Distinct
抑制（过滤掉）重复的数据项
distinct.png
Observable.just(1,2,1,1,2,1,1,2,3,4)
.distinct(new Function<Integer, Integer>() {
@Override
public Integer apply(@NonNull Integer integer) throws Exception {
return integer;
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.i(TAG, "accept: integer = " + integer);
}
});
08-24 17:26:19.198 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 1
08-24 17:26:19.199 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 2
08-24 17:26:19.199 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 3
08-24 17:26:19.199 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 4
-
ElementAt
只发射第N项数据 -
Filter
Filter操作符使用你指定的一个谓词函数测试数据项，只有通过测试的数据才会被发射。
举个例子,过滤整数序列中的奇数,只发射偶数
filter.png
Observable.interval(1,TimeUnit.SECONDS)
.filter(new Predicate<Long>() {
@Override
public boolean test(@NonNull Long aLong) throws Exception {
if (aLong % 2 == 0){
return true;
}
return false;
}
})
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.i(TAG, "accept: " + aLong);
}
});
08-24 17:30:57.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 0
08-24 17:30:59.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 2
08-24 17:31:01.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 4
08-24 17:31:03.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 6
08-24 17:31:05.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 8
08-24 17:31:07.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 10
-
First
只发射第一项（或者满足某个条件的第一项）数据 -
IgnoreElements
不发射任何数据，只发射Observable的终止通知 -
skip/take
skip
image.png
take
image.png
- skipLast/takeLast
image.png
image.png
-
Sample
定期发射Observable最近发射的数据项
image.png
RxJava将这个操作符实现为sample和throttleLast。
Observable.range(0,1000)
.sample(1, TimeUnit.MICROSECONDS)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.i(TAG, "accept: " + integer);
}
});
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 56
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 154
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 186
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 208
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 228
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 247
组合操作
- zip
通过一个函数将多个Observables的发射物结合到一起，基于这个函数的结果为每个结合体发射单个数据项。
Zip操作符返回一个Obversable，它使用这个函数按顺序结合两个或多个Observables发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据。
image.png
上代码:
Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Thread.sleep(1000);
e.onNext(1);
Thread.sleep(1000);
e.onNext(2);
Thread.sleep(1000);
e.onNext(3);
Thread.sleep(1000);
e.onNext(4);
Thread.sleep(1000);
e.onComplete();
}
}).subscribeOn(Schedulers.newThread());
Observable<String> stringObservable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
Thread.sleep(700);
e.onNext("A");
Thread.sleep(700);
e.onNext("B");
Thread.sleep(700);
e.onNext("C");
Thread.sleep(700);
e.onNext("D");
Thread.sleep(700);
e.onNext("E");
Thread.sleep(700);
e.onNext("F");
Thread.sleep(700);
e.onComplete();
}
}).subscribeOn(Schedulers.newThread());
Observable
.zip(stringObservable, integerObservable, new BiFunction<String, Integer, String>() {
@Override
public String apply(@NonNull String s, @NonNull Integer integer) throws Exception {
return s + integer;
}
})
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.i(TAG, "onSubscribe: ");
}
@Override
public void onNext(@NonNull String s) {
Log.i(TAG, "onNext: " +s);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
Log.i(TAG, "onComplete: ");
}
});
08-24 17:50:11.195 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: A1
08-24 17:50:12.196 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: B2
08-24 17:50:13.197 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: C3
08-24 17:50:14.198 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: D4
08-24 17:50:15.198 30117-30157/com.example.wty.learnrxjava I/MainActivity: onComplete:
错误处理
- Retry
如果原始Observable遇到错误(即接收到onError的时候触发)，重新订阅它期望它能正常终止
image.png
-
RetryWhen
retryWhen和retry类似，区别是，retryWhen将onError中的Throwable传递给一个函数，这个函数产生另一个Observable，retryWhen观察它的结果再决定是不是要重新订阅原始的Observable。如果这个Observable发射了一项数据，它就重新订阅，如果这个Observable发射的是onError通知，它就将这个通知传递给观察者然后终止。
Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Thread.sleep(1000);
e.onNext(1);
Thread.sleep(1000);
e.onNext(2);
Thread.sleep(1000);
e.onNext(3);
int value = 1 / 0;
Thread.sleep(1000);
e.onNext(4);
Thread.sleep(1000);
e.onComplete();
}
}).subscribeOn(Schedulers.newThread());
integerObservable.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {
if (throwable instanceof ArithmeticException)
return Observable.just(1);
else
return Observable.error(throwable);
}
});
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.i(TAG, "onSubscribe: ");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.i(TAG, "onNext: " + integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError: ", e);
}
@Override
public void onComplete() {
Log.i(TAG, "onComplete: ");
}
});
08-24 17:56:24.620 3895-3935/com.example.wty.learnrxjava I/MainActivity: onNext: 1
08-24 17:56:25.621 3895-3935/com.example.wty.learnrxjava I/MainActivity: onNext: 2
08-24 17:56:26.621 3895-3935/com.example.wty.learnrxjava I/MainActivity: onNext: 3
08-24 17:56:27.624 3895-3998/com.example.wty.learnrxjava I/MainActivity: onNext: 1
08-24 17:56:28.624 3895-3998/com.example.wty.learnrxjava I/MainActivity: onNext: 2
08-24 17:56:29.625 3895-3998/com.example.wty.learnrxjava I/MainActivity: onNext: 3
08-24 17:56:30.629 3895-4038/com.example.wty.learnrxjava I/MainActivity: onNext: 1
08-24 17:56:31.630 3895-4038/com.example.wty.learnrxjava I/MainActivity: onNext: 2
08-24 17:56:32.630 3895-4038/com.example.wty.learnrxjava I/MainActivity: onNext: 3
线程调度
-
SubscribeOn
指定Observable自身在哪个调度器上执行 -
ObserveOn
指定一个观察者在哪个调度器上观察这个Observable
