takeUtil
/**
* Returns an Observable that emits items emitted by the source Observable, checks the specified predicate
* for each item, and then completes when the condition is satisfied.
* <p>
* <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/takeUntil.p.png" alt="">
* <p>
* The difference between this operator and {@link #takeWhile(Predicate)} is that here, the condition is
* evaluated <em>after</em> the item is emitted.
*
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code takeUntil} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
*
* @param stopPredicate
* a function that evaluates an item emitted by the source Observable and returns a Boolean
* @return an Observable that first emits items emitted by the source Observable, checks the specified
* condition after each item, and then completes when the condition is satisfied.
* @see <a href="http://reactivex.io/documentation/operators/takeuntil.html">ReactiveX operators documentation: TakeUntil</a>
* @see Observable#takeWhile(Predicate)
* @since 1.1.0
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> takeUntil(Predicate<? super T> stopPredicate) {
ObjectHelper.requireNonNull(stopPredicate, "predicate is null");
return RxJavaPlugins.onAssembly(new ObservableTakeUntilPredicate<T>(this, stopPredicate));
}
核心逻辑代码
@Override
public void onNext(T t) {
if (!done) {
//一开始就发送一个onNext事件
downstream.onNext(t);
boolean b;
try {
//然后就开始判断条件是否满足
b = predicate.test(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
upstream.dispose();
onError(e);
return;
}
//直到条件满足，调用链才完成
if (b) {
done = true;
upstream.dispose();
downstream.onComplete();
}
}
}
takeUtil 是在检查条件之前发射数据，直到满足条件时（发射的数据包括不满足条件的和一个满足条件的），然后就发射完成了，不在发射后续的数据。所以发射的数据包括满足条件的那一个。
image.png
takeWhile
/**
* Returns an Observable that emits items emitted by the source ObservableSource so long as each item satisfied a
* specified condition, and then completes as soon as this condition is not satisfied.
* <p>
* <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/takeWhile.png" alt="">
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code takeWhile} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
*
* @param predicate
* a function that evaluates an item emitted by the source ObservableSource and returns a Boolean
* @return an Observable that emits the items from the source ObservableSource so long as each item satisfies the
* condition defined by {@code predicate}, then completes
* @see <a href="http://reactivex.io/documentation/operators/takewhile.html">ReactiveX operators documentation: TakeWhile</a>
* @see Observable#takeUntil(Predicate)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> takeWhile(Predicate<? super T> predicate) {
ObjectHelper.requireNonNull(predicate, "predicate is null");
return RxJavaPlugins.onAssembly(new ObservableTakeWhile<T>(this, predicate));
}
@Override
public void onNext(T t) {
if (done) {
return;
}
boolean b;
try {
//一开始就进行判断
b = predicate.test(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
upstream.dispose();
onError(e);
return;
}
//直到条件不满足时，调用链完成
if (!b) {
done = true;
upstream.dispose();
downstream.onComplete();
return;
}
downstream.onNext(t);
}
takeWhile是每次发射之前都判断条件是否满足，条件满足就不发射数据了( 发射的数据都是不满足条件的)。所以就不包含满足条件的那一个。
image.png
总结
所以takeUtil与takeWhile的区别就是 判断条件的时机。takeUtil是发射数据后在判断(先发射数据，判断满足条件时，后续就不再发射数据)，takeWhile是每次发射数据前判断(不满足条件就发射)
