RxJava takeUtil与takewhile的区别.md

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是每次发射数据前判断(不满足条件就发射)