RxJava2极速入门——Rxjava操作符详解之过滤操作符
- RxJava操作符——过滤操作符
-
- elementAt
-
- 直接index索引
- 直接索引并设置默认角标
- elementAt(long index, T defaultItem) 与 elementAt(long index) 区别
- frist
- last
- IngoreElements
- filter
- distinct
-
- 基本使用
- 自定义key使用
- debounce
- take
-
- count用法
- time用法
- takeLast
-
- count的用法
- time的用法
- time and count用法
- skip
-
- count用法
- time用法
- skipLast
-
- count用法
- time用法
- 总结
RxJava操作符——过滤操作符
清明时节雨纷纷，路上行人欲断魂。感慨一下，哪怕是祭奠先人，也别忘了学习，一周一更。正编开始：
在ReactiveX中转换操作时这样子描述的Operators that selectively emit items from a source Observable，其含义就是将需要发射的Observables通过使用Filtering Operators这种操作相关的函数修饰符过滤成真正需要发射的Observable。
过滤操作符常见分类如下：
elementAt
elementAt:emit only item n emitted by an Observable
elementAt只获取Observables中指定角标的Observable并发射。
在elementAt中有两种用法：
直接index索引
原理图如下：
从原理图可以得出角标索引遵循index为0作为起始去搜索指定的数据，然后当截取到数据后将它发射出去最后终止整个事件的订阅。
所以传入的index必须大于等于0，并且保证角标不能越界否则会出现运行异常。
示例代码以及部分相关源码如下：
private fun operatorsElementAt() {
var justIndexDispoable: Disposable? = null
Observable.rangeLong(1, 5).elementAt(0).subscribe(
getBaseLongMaybeObserver("operatorsElementAt-使用角标索引", {
justIndexDispoable = it
}, {
justIndexDispoable?.dispose()
})
)
}
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsElementAt-使用角标索引: onSubscribe
* com.ypz.rxjavademo I/operatorsElementAt-使用角标索引: onNext:value1
* 相关源码如下
* */
/**
* Returns a Maybe that emits the single item at a specified index in a sequence of emissions from
* this Observable or completes if this Observable signals fewer elements than index.
* @param index
* the zero-based index of the item to retrieve
* @return a Maybe that emits a single item: the item at the specified position in the sequence of
* those emitted by the source ObservableSource
* @throws IndexOutOfBoundsException
* if {@code index} is less than 0
* @see <a href="http://reactivex.io/documentation/operators/elementat.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: ElementAt</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Maybe<T> elementAt(long index) {
if (index < 0) {
throw new IndexOutOfBoundsException("index >= 0 required but it was " + index);
}
return RxJavaPlugins.onAssembly(new ObservableElementAtMaybe<T>(this, index));
}
public final class ObservableElementAtMaybe<T> extends Maybe<T> implements FuseToObservable<T> {
final ObservableSource<T> source;
final long index;
public ObservableElementAtMaybe(ObservableSource<T> source, long index) {
this.source = source;
this.index = index;
}
tic final class ElementAtObserver<T> implements Observer<T>, Disposable {}
}
从示例代码以及源码可以得出 elementAt(long index) 这一用法得益于ObservableElementAtMaybe的实现，其通过继承Maybe并实现FuseToObservable，在通过ElementAtObserver的实现，注意这里的ElementAtObserver既为观察者也为被观察者一个中转作用，通过这种作用过滤出相应角标Observable进行发射最后取消订阅事件的
直接索引并设置默认角标
原理图如下:
从原理图可得当无法索引到相应角标的Observable，则通过所有defaultIndex去寻找相应的Observable。最终发射出去
示例代码以及相关源码如下:
private fun operatorsElementAt() {
var usedDefinedIndexDispoable: Disposable? = null
Observable.rangeLong(1, 5).elementAt(6, 3).subscribe(
getBaseLongSingleObserver("operatorsElementAt-默认角标用法",
{ usedDefinedIndexDispoable = it },
{ usedDefinedIndexDispoable?.dispose() })
)
}
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsElementAt-默认角标用法: onSubscribe
* com.ypz.rxjavademo I/operatorsElementAt-默认角标用法: onNext:value3
* 相关源码如下
* */
/**
* Returns a Single that emits the item found at a specified index in a sequence of emissions from
* this Observable, or a default item if that index is out of range.
* @param index
* the zero-based index of the item to retrieve
* @param defaultItem
* the default item
* @return a Single that emits the item at the specified position in the sequence emitted by the source
* ObservableSource, or the default item if that index is outside the bounds of the source sequence
* @throws IndexOutOfBoundsException
* if {@code index} is less than 0
* @see <a href="http://reactivex.io/documentation/operators/elementat.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: ElementAt</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Single<T> elementAt(long index, T defaultItem) {
if (index < 0) {
throw new IndexOutOfBoundsException("index >= 0 required but it was " + index);
}
ObjectHelper.requireNonNull(defaultItem, "defaultItem is null");
return RxJavaPlugins.onAssembly(new ObservableElementAtSingle<T>(this, index, defaultItem));
}
public final class ObservableElementAtSingle<T> extends Single<T> implements FuseToObservable<T> {
final ObservableSource<T> source;
final long index;
final T defaultValue;
public ObservableElementAtSingle(ObservableSource<T> source, long index, T defaultValue) {
this.source = source;
this.index = index;
this.defaultValue = defaultValue;
}
@Override
public void subscribeActual(SingleObserver<? super T> t) {
source.subscribe(new ElementAtObserver<T>(t, index, defaultValue));
}
@Override
public Observable<T> fuseToObservable() {
return RxJavaPlugins.onAssembly(new ObservableElementAt<T>(source, index, defaultValue, true));
}
static final class ElementAtObserver<T> implements Observer<T>, Disposable {
/*内部源码省略*/
}
}
从示例代码以及源码可以得出 elementAt(long index, T defaultItem) 这一用法得益于ObservableElementAtSingle的实现，其通过继承single并实现FuseToObservable，在通过ElementAtObserver的实现，注意这里的ElementAtObserver既为观察者也为被观察者一个中转作用，通过这种作用过滤出相应角标Observable进行发射最后取消订阅事件的
elementAt(long index, T defaultItem) 与 elementAt(long index) 区别
elementAt(long index)底层实现基于ObservableElementAtMaybe以及ObservableElementAtMaybe内部的ElementAtObserver，返回的是Maybe，ObservableElementAtMaybe继承Maybe。
elementAt(long index, T defaultItem)底层实现基于ObservableElementAtSingle以及ObservableElementAtSingle内部的ElementAtObserver，返回的是Single，ObservableElementAtMaybe继承Single。
两者都实现了FuseToObservable接口。
frist
first:emit only the first item (or the first item that meets some condition) emitted by an Observable
从Observables中发射第一个(符合条件的元素)
原理图如下:
从原理图中可以看到first与**elementAt(long index, T defaultItem)**十分相似，他们是否就是一致的呢请看接下来的代码。
示例代码以及相关源码如下:
private fun operatorsFirst() {
var disposable: Disposable? = null
Observable.rangeLong(1, 5).first(1).subscribe(
getBaseLongSingleObserver("operatorsFirst",
{ disposable = it },
{ disposable?.dispose() }))
}
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsFirst: onSubscribe
* com.ypz.rxjavademo I/operatorsFirst: onNext:value1
* 源码如下:
* */
/**
* Returns a Single that emits only the very first item emitted by the source ObservableSource, or a default item
* if the source ObservableSource completes without emitting any items.
*
* @param defaultItem
* the default item to emit if the source ObservableSource doesn't emit anything
* @return the new Single instance
* @see <a href="http://reactivex.io/documentation/operators/first.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: First</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Single<T> first(T defaultItem) {
return elementAt(0L, defaultItem);
}
结合示例代码以及源码得出frist其实就是elementAt(long index, T defaultItem)，就是换烫不换药，但是first默认是获取第一项起始角标。
last
last:emit only the last item (or the last item that meets some condition) emitted by an Observable
从Observables中发射最后一个(符合条件的元素)
原理图如下:
private fun operatorsLast() {
var disposable: Disposable? = null
Observable.rangeLong(1, 5).last(1).subscribe(
getBaseLongSingleObserver("operatorsLast",
{ disposable = it },
{ disposable?.dispose() }))
}
/**
* 运行结果如下:
* com.ypz.rxjavademo I/operatorsLast: onSubscribe
* com.ypz.rxjavademo I/operatorsLast: onNext:value5
* 相关源码如下:
* */
/**
* Returns a Single that emits only the last item emitted by this Observable, or a default item
* if this Observable completes without emitting any items.
* @param defaultItem
* the default item to emit if the source ObservableSource is empty
* @return a Single that emits only the last item emitted by the source ObservableSource, or a default item
* if the source ObservableSource is empty
* @see <a href="http://reactivex.io/documentation/operators/last.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: Last</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Single<T> last(T defaultItem) {
ObjectHelper.requireNonNull(defaultItem, "defaultItem is null");
return RxJavaPlugins.onAssembly(new ObservableLastSingle<T>(this, defaultItem));
}
public final class ObservableLastSingle<T> extends Single<T> {
final ObservableSource<T> source;
final T defaultItem;
public ObservableLastSingle(ObservableSource<T> source, T defaultItem) {
this.source = source;
this.defaultItem = defaultItem;
}
// TODO fuse back to Observable
@Override
protected void subscribeActual(SingleObserver<? super T> observer) {
source.subscribe(new LastObserver<T>(observer, defaultItem));
}
static final class LastObserver<T> implements Observer<T>, Disposable {
//部分源码省略
}
}
结合示例代码以及源码可以得出last的实现是基于ObservableLastSingle的实现，而ObservableLastSingle则通过继承Single以及内部LastObserver的实现。
IngoreElements
IngoreElements:do not emit any items from an Observable but mirror its termination notification
原理图如下:
结合原理图可以得出:不发射任何数据只关注与数据是否发送完成或者发送出现异常中断
示例代码以及源码如下:
private fun operatorsIgnoreElements() {
Observable.rangeLong(1, 5).ignoreElements().subscribe(
{ logIMessage("operatorsIgnoreElements", "onComplete") },
{
logIMessage("operatorsIgnoreElements", "errorMessage:${it.message
?: "unkown error"}")
})
}
/**
* 运行结果如下:
* com.ypz.rxjavademo I/operatorsIgnoreElements: onComplete
* 相关源码如下:
* */
/**
* Ignores all items emitted by the source ObservableSource and only calls {@code onComplete} or {@code onError}.
*
* @return the new Completable instance
* @see <a href="http://reactivex.io/documentation/operators/ignoreelements.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: IgnoreElements</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Completable ignoreElements() {
return RxJavaPlugins.onAssembly(new ObservableIgnoreElementsCompletable<T>(this));
}
public final class ObservableIgnoreElementsCompletable<T> extends Completable implements FuseToObservable<T> {
final ObservableSource<T> source;
public ObservableIgnoreElementsCompletable(ObservableSource<T> source) {
this.source = source;
}
@Override
public void subscribeActual(final CompletableObserver t) {
source.subscribe(new IgnoreObservable<T>(t));
}
@Override
public Observable<T> fuseToObservable() {
return RxJavaPlugins.onAssembly(new ObservableIgnoreElements<T>(source));
}
static final class IgnoreObservable<T> implements Observer<T>, Disposable {
@Override
public void onNext(T v) {
// deliberately ignored
}
}
}
结合示例代码以及源码可以看到IngoreElements底层是是基于ObservableIgnoreElementsCompletable的实现，而ObservableIgnoreElementsCompletable通过继承Completable并实现FuseToObservable接口，而其内部的IgnoreObservable这一类对onNext实现是一个nothing to do做法，专注于onComple以及onError的处理所实现的一个结果。
filter
filter:emit only those items from an Observable that pass a predicate test
原理图如下：
结合原理图可得：filter是将Observables进行条件过滤然后发射出去。
示例代码以及源码如下:
private fun operatorsFilter() {
Observable.rangeLong(1, 10).
filter { (it % 2 == 0L) }.
subscribe(ShowMessageBaseObserver<Long>("operatorsFilter"))
}
/**
* 运行结果如下:
* com.ypz.rxjavademo I/operatorsFilter: onSubscribe
* com.ypz.rxjavademo I/operatorsFilter: onNext:value2
* com.ypz.rxjavademo I/operatorsFilter: onNext:value4
* com.ypz.rxjavademo I/operatorsFilter: onNext:value6
* com.ypz.rxjavademo I/operatorsFilter: onNext:value8
* com.ypz.rxjavademo I/operatorsFilter: onNext:value10
* com.ypz.rxjavademo I/operatorsFilter: onComplete
* 源码如下:
* */
/**
* Filters items emitted by an ObservableSource by only emitting those that satisfy a specified predicate.
*
* @param predicate
* a function that evaluates each item emitted by the source ObservableSource, returning {@code true}
* if it passes the filter
* @return an Observable that emits only those items emitted by the source ObservableSource that the filter
* evaluates as {@code true}
* @see <a href="http://reactivex.io/documentation/operators/filter.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: Filter</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> filter(Predicate<? super T> predicate) {
ObjectHelper.requireNonNull(predicate, "predicate is null");
return RxJavaPlugins.onAssembly(new ObservableFilter<T>(this, predicate));
}
public final class ObservableFilter<T> extends AbstractObservableWithUpstream<T, T> {
final Predicate<? super T> predicate;
public ObservableFilter(ObservableSource<T> source, Predicate<? super T> predicate) {
super(source);
this.predicate = predicate;
}
@Override
public void subscribeActual(Observer<? super T> observer) {
source.subscribe(new FilterObserver<T>(observer, predicate));
}
static final class FilterObserver<T> extends BasicFuseableObserver<T, T> {
final Predicate<? super T> filter;
FilterObserver(Observer<? super T> actual, Predicate<? super T> filter) {
super(actual);
this.filter = filter;
}
@Override
public void onNext(T t) {
if (sourceMode == NONE) {
boolean b;
try {
b = filter.test(t);
} catch (Throwable e) {
fail(e);
return;
}
if (b) {
downstream.onNext(t);
}
} else {
downstream.onNext(null);
}
}
@Nullable
@Override
public T poll() throws Exception {
for (;;) {
T v = qd.poll();
if (v == null || filter.test(v)) {
return v;
}
}
}
}
}
public interface Predicate<T> {
/**
* Test the given input value and return a boolean.
* @param t the value
* @return the boolean result
* @throws Exception on error
*/
boolean test(@NonNull T t) throws Exception;
}
结合示例代码以及源码得出：filter底层是基于ObservableFilter的实现，通过对Predicate接口的实现，在onNext事件中反复复用Predicate进行过滤发送。
distinct
distinct:suppress duplicate items emitted by an Observable
过滤重复的元素保证发射的数据是唯一的
基本使用
其原理图如下:
示例代码以及源码如下:
private fun operatorsDistinct() {
Observable.just(1L, 2L, 3L, 1L, 2L).distinct()
.subscribe(ShowMessageBaseObserver<Long>("operatorsDistinct-Easy用法"))
}
/**
* 运行结果：
* com.ypz.rxjavademo I/operatorsDistinct-Easy用法: onSubscribe
* com.ypz.rxjavademo I/operatorsDistinct-Easy用法: onNextValue:1
* com.ypz.rxjavademo I/operatorsDistinct-Easy用法: onNextValue:2
* com.ypz.rxjavademo I/operatorsDistinct-Easy用法: onNextValue:3
* com.ypz.rxjavademo I/operatorsDistinct-Easy用法: onComplete
* 相关源码如下：
* */
/**
* Returns an Observable that emits all items emitted by the source ObservableSource that are distinct
* based on {@link Object#equals(Object)} comparison.
* Customizing the retention policy can happen only by providing a custom {@link java.util.Collection} implementation
* to the {@link #distinct(Function, Callable)} overload.
*
* @return an Observable that emits only those items emitted by the source ObservableSource that are distinct from
* each other
* @see #distinct(Function)
* @see #distinct(Function, Callable)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> distinct() {
return distinct(Functions.identity(), Functions.createHashSet());
}
public static <T> Callable<Set<T>> createHashSet() {
return (Callable)HashSetCallable.INSTANCE;
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <K> Observable<T> distinct(Function<? super T, K> keySelector, Callable<? extends Collection<? super K>> collectionSupplier) {
ObjectHelper.requireNonNull(keySelector, "keySelector is null");
ObjectHelper.requireNonNull(collectionSupplier, "collectionSupplier is null");
return RxJavaPlugins.onAssembly(new ObservableDistinct<T, K>(this, keySelector, collectionSupplier));
}
public final class ObservableDistinct<T, K> extends AbstractObservableWithUpstream<T, T> {
final Function<? super T, K> keySelector;
final Callable<? extends Collection<? super K>> collectionSupplier;
public ObservableDistinct(ObservableSource<T> source, Function<? super T, K> keySelector, Callable<? extends Collection<? super K>> collectionSupplier) {
super(source);
this.keySelector = keySelector;
this.collectionSupplier = collectionSupplier;
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
Collection<? super K> collection;
try {
collection = ObjectHelper.requireNonNull(collectionSupplier.call(), "The collectionSupplier returned a null collection. Null values are generally not allowed in 2.x operators and sources.");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
EmptyDisposable.error(ex, observer);
return;
}
source.subscribe(new DistinctObserver<T, K>(observer, keySelector, collection));
}
static final class DistinctObserver<T, K> extends BasicFuseableObserver<T, T> {
final Collection<? super K> collection;
final Function<? super T, K> keySelector;
DistinctObserver(Observer<? super T> actual, Function<? super T, K> keySelector, Collection<? super K> collection) {
super(actual);
this.keySelector = keySelector;
this.collection = collection;
}
@Override
public void onNext(T value) {
if (done) {
return;
}
if (sourceMode == NONE) {
K key;
boolean b;
try {
key = ObjectHelper.requireNonNull(keySelector.apply(value), "The keySelector returned a null key");
b = collection.add(key);
} catch (Throwable ex) {
fail(ex);
return;
}
if (b) {
downstream.onNext(value);
}
} else {
downstream.onNext(null);
}
}
}
}
结合示例代码以及源码得出：
首先在distinct()中，其实是使用到自定义Key一种封装的实现，通过内部已经实现的Functions.createHashSet()去创建一个Hash，然后利用这一个HashSet对key进行比较是否数据重复了，然后去重发射数据，注意这种使用下value 等价于 key。
自定义key使用
原理图如下:
示例代码如下
private fun distinctKey() {
var times = 1L
Observable
.just(1L, 2L, 3L, 1L, 2L)
.distinct {
times += 1
times
}
.subscribe(ShowMessageBaseObserver<Long>("operatorsDistinct-Key用法"))
}
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsDistinct-Key用法: onSubscribe
* com.ypz.rxjavademo I/operatorsDistinct-Key用法: onNextValue:1
* com.ypz.rxjavademo I/operatorsDistinct-Key用法: onNextValue:2
* com.ypz.rxjavademo I/operatorsDistinct-Key用法: onNextValue:3
* com.ypz.rxjavademo I/operatorsDistinct-Key用法: onNextValue:1
* com.ypz.rxjavademo I/operatorsDistinct-Key用法: onNextValue:2
* com.ypz.rxjavademo I/operatorsDistinct-Key用法: onComplete
* */
结合示例代码可以看到当使用Function去实现自定义Key的时候，因为key已经不在将onNext事件中Value看作为Key所以也就没有达到类似简单用法的去重效果，但是其去重是已经实现了。可见在distinct的去重效果是由key中value对比所得，实现这种自定义Key的用法时候必须使用统一的校准规则避免数据去重校检出错达不到预期效果。
debounce
debounce:suppress duplicate items emitted by an Observable
过滤发射数据过快的数据
原理图如下:
示例代码以及源码如下:
private fun operatorsDebounce() {
Observable.create(ObservableOnSubscribe<Long> { emitter ->
if (emitter.isDisposed) return@ObservableOnSubscribe
for (index in 1L..7L) {
emitter.onNext(index)
Thread.sleep(index * 100L)
}
emitter.onComplete()
})
.debounce(500, TimeUnit.MILLISECONDS)
.subscribe(ShowMessageBaseObserver<Long>("operatorsDebounce"))
}
/**
* 运行结果如下:
* com.ypz.rxjavademo I/operatorsDebounce: onSubscribe
* com.ypz.rxjavademo I/operatorsDebounce: onNext:value6
* com.ypz.rxjavademo I/operatorsDebounce: onNext:value7
* com.ypz.rxjavademo I/operatorsDebounce: onComplete
* 源码如下:
* */
/**
* Returns an Observable that mirrors the source ObservableSource, except that it drops items emitted by the
* source ObservableSource that are followed by newer items before a timeout value expires. The timer resets on
* each emission.
*
* @param timeout
* the length of the window of time that must pass after the emission of an item from the source
* ObservableSource in which that ObservableSource emits no items in order for the item to be emitted by the
* resulting ObservableSource
* @param unit
* the unit of time for the specified {@code timeout}
* @return an Observable that filters out items from the source ObservableSource that are too quickly followed by
* newer items
* @see <a href="http://reactivex.io/documentation/operators/debounce.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: Debounce</a>
* @see #throttleWithTimeout(long, TimeUnit)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.COMPUTATION)
public final Observable<T> debounce(long timeout, TimeUnit unit) {
return debounce(timeout, unit, Schedulers.computation());
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> debounce(long timeout, TimeUnit unit, Scheduler scheduler) {
ObjectHelper.requireNonNull(unit, "unit is null");
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableDebounceTimed<T>(this, timeout, unit, scheduler));
}
结合示例代码以及源码得出：
这种用法下其底层基于ObservableDebounceTimed的实现，其实现通过利用TimeUnit 进行时间控制，每发射一个数据后进入一个冷却期间，冷却期间不接收上游的数据发射，从而过滤发射数据过快的事件。
take
take:emit only the first n items emitted by an Observable
count用法
原理图如下:
示例代码以及源码如下:
private fun operatorsTakeCount() {
Observable.
rangeLong(1L, 5L).take(3).
subscribe(ShowMessageBaseObserver<Long>("operatorsTake_count"))
}
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsTake_count: onSubscribe
* com.ypz.rxjavademo I/operatorsTake_count: onNext:value1
* com.ypz.rxjavademo I/operatorsTake_count: onNext:value2
* com.ypz.rxjavademo I/operatorsTake_count: onNext:value3
* com.ypz.rxjavademo I/operatorsTake_count: onComplete
* 相关源码如下:
* */
/**
* Returns an Observable that emits only the first {@code count} items emitted by the source ObservableSource. If the source emits fewer than
* {@code count} items then all of its items are emitted.
* @param count
* the maximum number of items to emit
* @return an Observable that emits only the first {@code count} items emitted by the source ObservableSource, or
* all of the items from the source ObservableSource if that ObservableSource emits fewer than {@code count} items
* @see <a href="http://reactivex.io/documentation/operators/take.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: Take</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> take(long count) {
if (count < 0) {
throw new IllegalArgumentException("count >= 0 required but it was " + count);
}
return RxJavaPlugins.onAssembly(new ObservableTake<T>(this, count));
}
public final class ObservableTake<T> extends AbstractObservableWithUpstream<T, T> {
final long limit;
public ObservableTake(ObservableSource<T> source, long limit) {
super(source);
this.limit = limit;
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
source.subscribe(new TakeObserver<T>(observer, limit));
}
static final class TakeObserver<T> implements Observer<T>, Disposable {
}
}
结合示例代码以及源码得出：
count的单纯是通过对count进行累加，注意是从1开始。当达到目标count值得时候中断发射。
time用法
原理图如下：
示例代码以及源码如下:
private fun operatorsTakeTime() =
Observable.
interval(1, TimeUnit.SECONDS).
take(3, TimeUnit.SECONDS).
subscribe(ShowMessageBaseObserver<Long>("operatorsTake_time"))
/**
* 运行结果如下:
* 第一次运行：
* com.ypz.rxjavademo I/operatorsTake_time: onSubscribe
* com.ypz.rxjavademo I/operatorsTake_time: onNext:value0
* com.ypz.rxjavademo I/operatorsTake_time: onNext:value1
* com.ypz.rxjavademo I/operatorsTake_time: onComplete
* 第二次运行：
* com.ypz.rxjavademo I/operatorsTake_time: onSubscribe
* com.ypz.rxjavademo I/operatorsTake_time: onNext:value0
* com.ypz.rxjavademo I/operatorsTake_time: onNext:value1
* com.ypz.rxjavademo I/operatorsTake_time: onNext:value2
* com.ypz.rxjavademo I/operatorsTake_time: onComplete
*
* */
/**
* Returns an Observable that emits those items emitted by source ObservableSource before a specified time runs
* out.
*
* @param time
* the length of the time window
* @param unit
* the time unit of {@code time}
* @return an Observable that emits those items emitted by the source ObservableSource before the time runs out
* @see <a href="http://reactivex.io/documentation/operators/take.html" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" target="_blank" rel="external nofollow" >ReactiveX operators documentation: Take</a>
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> take(long time, TimeUnit unit) {
return takeUntil(timer(time, unit));
}
结合示例代码以及源码得出：
这一用法是基于timer以及takeUntil封装实现所得，而takeUntil属于条件与布尔操作符将在下一篇详解。
这种用法的核心思维就是，获取起始某个时间段内的数据，当超过时间片段时则不在获取数据，将获取到数据进行发送，当时间段已经超过的时间会自动中断事件的发送。
从第一次和第二次运行结果不一致更加验证了这个道理，为什么第一次发送的事件少了一次，因为程序并发运行的时间段并不是超精准的会存在微妙的误差所致。
takeLast
take:emit only the final n items emitted by an Observable
count的用法
原理图如下:
示例代码以及源码如下:
private fun operatorsTakeLastCount() =
Observable.
rangeLong(1L, 5L).
takeLast(2).
subscribe(ShowMessageBaseObserver("operatorsTakeLast_count"))
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsTakeLast_count: onSubscribe
* com.ypz.rxjavademo I/operatorsTakeLast_count: onNext:value4
* com.ypz.rxjavademo I/operatorsTakeLast_count: onNext:value5
* com.ypz.rxjavademo I/operatorsTakeLast_count: onComplete
*
* */
/**
* Returns an Observable that emits at most the last {@code count} items emitted by the source ObservableSource. If the source emits fewer than
* {@code count} items then all of its items are emitted.
*
* @param count
* the maximum number of items to emit from the end of the sequence of items emitted by the source
* ObservableSource
* @return an Observable that emits at most the last {@code count} items emitted by the source ObservableSource
* @throws IndexOutOfBoundsException
* if {@code count} is less than zero
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> takeLast(int count) {
if (count < 0) {
throw new IndexOutOfBoundsException("count >= 0 required but it was " + count);
} else
if (count == 0) {
return RxJavaPlugins.onAssembly(new ObservableIgnoreElements<T>(this));
} else
if (count == 1) {
return RxJavaPlugins.onAssembly(new ObservableTakeLastOne<T>(this));
}
return RxJavaPlugins.onAssembly(new ObservableTakeLast<T>(this, count));
}
结合示例代码以及源码得出：
从源码可以得到当count 的值大小返回相应的结果，
当count :<0时，返回运行异常
当count:0时，返回ObservableIgnoreElements
当count :1时，返回ObservableTakeLastOne
当count:>1时，返回ObservableTakeLast
ObservableTakeLastOne与ObservableTakeLast区别在于对onNext以及一些事件实现方式有些区别，其本质都是截取最后指定位置的元素进行发射。
time的用法
原理图如下:
示例代码以及源码如下:
private fun operatorsTakeLastTime() =
Observable.
intervalRange(0, 6, 0, 1, TimeUnit.SECONDS).
takeLast(3, TimeUnit.SECONDS).
subscribe(ShowMessageBaseObserver<Long>("operatorsTakeLast_time"))
/**
* 运行结果如下:
* com.ypz.rxjavademo I/operatorsTakeLast_time: onSubscribe
* com.ypz.rxjavademo I/operatorsTakeLast_time: onNext:value3
* com.ypz.rxjavademo I/operatorsTakeLast_time: onNext:value4
* com.ypz.rxjavademo I/operatorsTakeLast_time: onNext:value5
* com.ypz.rxjavademo I/operatorsTakeLast_time: onComplete
*
* */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.TRAMPOLINE)
public final Observable<T> takeLast(long time, TimeUnit unit) {
return takeLast(time, unit, Schedulers.trampoline(), false, bufferSize());
}
public final Observable<T> takeLast(long count, long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(unit, "unit is null");
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
if (count < 0) {
throw new IndexOutOfBoundsException("count >= 0 required but it was " + count);
}
return RxJavaPlugins.onAssembly(
new ObservableTakeLastTimed<T>(this, count, time, unit, scheduler, bufferSize, delayError));
}
结合示例代码以及源码得出：
time用法在于最后一个事件发送时，在发送前的一段时间段内所产生的事件会按照顺序过滤出来并按照顺序发射。
从源码可得出其实time用法就是基于time and count的组合二次封装，只不过是count设定为1。
time and count用法
原理图如下:
示例代码以及源码如下:
private fun operatorsTakeLastCountTime() =
Observable.
intervalRange(0, 6, 0, 1, TimeUnit.SECONDS).
takeLast(2, 3, TimeUnit.SECONDS).
subscribe(ShowMessageBaseObserver("operatorsTakeLast_count"))
/**
* 运行结果如下
* com.ypz.rxjavademo I/operatorsTakeLast_count: onSubscribe
* com.ypz.rxjavademo I/operatorsTakeLast_count: onNext:value4
* com.ypz.rxjavademo I/operatorsTakeLast_count: onNext:value5
* com.ypz.rxjavademo I/operatorsTakeLast_count: onComplete
* 相关源码如下：
* */
public final Observable<T> takeLast(long count, long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(unit, "unit is null");
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
if (count < 0) {
throw new IndexOutOfBoundsException("count >= 0 required but it was " + count);
}
return RxJavaPlugins.onAssembly(
new ObservableTakeLastTimed<T>(this, count, time, unit, scheduler, bufferSize, delayError));
}
结合示例代码以及源码得出：
count大小可以看作是一个window的size。而time则为观测收集数据的时间片段。
cout的时间片段等于time，则window可发射的size等于count的大小。
cout的时间片段小于time，则window可发射的size等于count的大小。
cout的时间片段大于time，则window可发射的size小于count的大小，并且会丢失数据。
skip
take:suppress the first n items emitted by an Observable
跳过指定条件内，第n位前产生的数据，采集n位后的数据
count用法
原理图如下:
示例代码以及源码如下:
private fun operatorsSkip() {
Observable.rangeLong(1, 5).skip(3).subscribe(
ShowMessageBaseObserver<Long>("operatorsSkip-count")
)
}
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsSkip-count: onSubscribe
* com.ypz.rxjavademo I/operatorsSkip-count: onNext:value4
* com.ypz.rxjavademo I/operatorsSkip-count: onNext:value5
* com.ypz.rxjavademo I/operatorsSkip-count: onComplete
* 源码如下:
* */
/**
* Returns an Observable that skips the first {@code count} items emitted by the source ObservableSource and emits
* the remainder.
*
* @param count
* the number of items to skip
* @return an Observable that is identical to the source ObservableSource except that it does not emit the first
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> skip(long count) {
if (count <= 0) {
return RxJavaPlugins.onAssembly(this);
}
return RxJavaPlugins.onAssembly(new ObservableSkip<T>(this, count));
public final class ObservableSkip<T> extends AbstractObservableWithUpstream<T, T> {
final long n;
public ObservableSkip(ObservableSource<T> source, long n) {
super(source);
this.n = n;
}
static final class SkipObserver<T> implements Observer<T>, Disposable {
final Observer<? super T> downstream;
long remaining;
Disposable upstream;
SkipObserver(Observer<? super T> actual, long n) {
this.downstream = actual;
this.remaining = n;
}
@Override
public void onNext(T t) {
if (remaining != 0L) {
remaining--;
} else {
downstream.onNext(t);
}
}
}
}
结合示例代码以及源码得出：
count用法：通过ObservableSkip的实现所得，其中跳过的关键在于ObservableSkip中SkipObserver的onNext的实现。通过累减从而达到skip的作用，并按照事件顺序发射数据。
time用法
原理图如下:
示例代码以及源码如下:
private fun operatorsSkipTime() {
Observable.interval(1, TimeUnit.SECONDS).take(5).skip(3, TimeUnit.SECONDS)
.subscribe(ShowMessageBaseObserver<Long>("operatorsSkip_time"))
}
/**
* 运行结果如下：
* com.ypz.rxjavademo I/operatorsSkip_time: onSubscribe
* com.ypz.rxjavademo I/operatorsSkip_time: onNext:value2
* com.ypz.rxjavademo I/operatorsSkip_time: onNext:value3
* com.ypz.rxjavademo I/operatorsSkip_time: onNext:value4
* com.ypz.rxjavademo I/operatorsSkip_time: onComplete
* 源码如下:
* */
/**
* Returns an Observable that skips values emitted by the source ObservableSource before a specified time window
* elapses.
*
* @param time
* the length of the time window to skip
* @param unit
* the time unit of {@code time}
* @return an Observable that skips values emitted by the source ObservableSource before the time window defined
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.COMPUTATION)
public final Observable<T> skip(long time, TimeUnit unit) {
return skipUntil(timer(time, unit));
}
结合示例代码以及源码得出：
time用法底层是基于条件布尔操作符skipUntil的实现所得，其中通过内部封装的timer实现，从而达到在time片段不在观测数据，当time片段失效时将观测所得的数据按照事件顺序进行发射。
skipLast
take:emit only the final n items emitted by an Observable
count用法
原理图如下:
示例代码以及源码如下:
private fun operatorsSkipLast() {
Observable.rangeLong(1, 5).skipLast(3)
.subscribe(ShowMessageBaseObserver<Long>("operatorsSkipLast-count"))
}
/**
* 运行结果如下:
* com.ypz.rxjavademo I/operatorsSkipLast-count: onSubscribe
* com.ypz.rxjavademo I/operatorsSkipLast-count: onNext:value1
* com.ypz.rxjavademo I/operatorsSkipLast-count: onNext:value2
* com.ypz.rxjavademo I/operatorsSkipLast-count: onComplete
* 源码如下:
* */
/**
* Returns an Observable that drops a specified number of items from the end of the sequence emitted by the
* source ObservableSource.
*
* @param count
* number of items to drop from the end of the source sequence
* @return an Observable that emits the items emitted by the source ObservableSource except for the dropped ones
* at the end
* @throws IndexOutOfBoundsException
* if {@code count} is less than zero
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> skipLast(int count) {
if (count < 0) {
throw new IndexOutOfBoundsException("count >= 0 required but it was " + count);
}
if (count == 0) {
return RxJavaPlugins.onAssembly(this);
}
return RxJavaPlugins.onAssembly(new ObservableSkipLast<T>(this, count));
}
public final class ObservableSkipLast<T> extends AbstractObservableWithUpstream<T, T> {
final int skip;
public ObservableSkipLast(ObservableSource<T> source, int skip) {
super(source);
this.skip = skip;
}
@Override
public void subscribeActual(Observer<? super T> observer) {
source.subscribe(new SkipLastObserver<T>(observer, skip));
}
static final class SkipLastObserver<T> extends ArrayDeque<T> implements Observer<T>, Disposable {
private static final long serialVersionUID = -3807491841935125653L;
final Observer<? super T> downstream;
final int skip;
Disposable upstream;
SkipLastObserver(Observer<? super T> actual, int skip) {
super(skip);
this.downstream = actual;
this.skip = skip;
}
@Override
public void onSubscribe(Disposable d) {
if (DisposableHelper.validate(this.upstream, d)) {
this.upstream = d;
downstream.onSubscribe(this);
}
}
@Override
public void onNext(T t) {
if (skip == size()) {
downstream.onNext(poll());
}
offer(t);
}
}
}
结合示例代码以及源码得出：
count用法：
当count<0时，会产生IndexOutOfBoundsException异常。
当count=1时，会直接返回原事件流。
当count>1时，count==size()时候，会触发downstream.onNext(poll())也就是直接事件流直接回调用onComplete()事件。其次通过使用offer将事件看成一个集合，然后利用**java.util.offer(T t)**这一方法实现跳过指定最后一位至最后一位前n位的事件发射
time用法
原理图如下:
示例代码以及源码如下:
private fun operatorsSkipLastTime() {
Observable.interval(1, TimeUnit.SECONDS).take(6)
.skipLast(4, TimeUnit.SECONDS).subscribe(
ShowMessageBaseObserver<Long>("operatorsSkipLast-time"))
}
/**
* 运行结果如下:
* com.ypz.rxjavademo I/operatorsSkipLast-time: onSubscribe
* com.ypz.rxjavademo I/operatorsSkipLast-time: onNext:value0
* com.ypz.rxjavademo I/operatorsSkipLast-time: onNext:value1
* com.ypz.rxjavademo I/operatorsSkipLast-time: onComplete
* 相关源码如下:
* */
/**
* Returns an Observable that drops items emitted by the source ObservableSource during a specified time window
* (defined on a specified scheduler) before the source completes.
*
* @param time
* the length of the time window
* @param unit
* the time unit of {@code time}
* @param scheduler
* the scheduler used as the time source
* @param delayError
* if true, an exception signalled by the current Observable is delayed until the regular elements are consumed
* by the downstream; if false, an exception is immediately signalled and all regular elements dropped
* @param bufferSize
* the hint about how many elements to expect to be skipped
* @return an Observable that drops those items emitted by the source ObservableSource in a time window before the
* source completes defined by {@code time} and {@code scheduler}
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> skipLast(long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(unit, "unit is null");
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
// the internal buffer holds pairs of (timestamp, value) so double the default buffer size
int s = bufferSize << 1;
return RxJavaPlugins.onAssembly(new ObservableSkipLastTimed<T>(this, time, unit, scheduler, s, delayError));
}
结合示例代码以及源码得出：
其实skipLast time用法就是将发射事件onComple调用前那段时间处于一个冷却期间的时间片段不在收集数据，未触发冷却期间的时间片段一直收集数据并按照顺序发射出去。
总结
elementAt两种用法：elementAt(long index, T defaultItem) 与 elementAt(long index)分别继承不同的Single、Maybe的区别以及实现方式
frist基于elementAt(long index, T defaultItem) 实现，专注于获取第一项数据
last基于ObservableLastSingle实现方式，专注于获取最后一项数据
IngoreElements基于ObservableIgnoreElementsCompletable实现，专注于事件是否发送完成以及是否触发错误
filter基于ObservableFilter以及Predicate的实现，达到自定义过滤事件的条件器作用
distinct去重的作用有两种实现方式，如果简单用法不适用建议使用自定义Key的用法从而达到数据去重的结果
debounce作用于过滤数据发射过块的作用基于ObservableDebounceTimed的实现
take两种使用方式count和time；每一种方式实现有区别一种适应于没有时间段的观测一种适用于时间段的观测
takeLast三种使用方式count和time、count与time组合；每一种方式实现有区别一种适应于没有时间段的观测一种适用于时间段的观测剩下一种适用于时间段内观察数据数据大小有关
skip两种使用方式count和time；每一种方式实现有区别一种适应于没有时间段的观测一种适用于时间段的观测
skipLast两种使用方式count和time；每一种方式实现有区别一种适应于没有时间段的观测一种适用于时间段的观测
需要区分出那种场景使用哪一种过滤方式，以及每一种实现底层类为什么。作用是什么。
点我代码传送欢迎star