takeUntil操作符探究

之前看了下RxLifecycle的实现，短短几行代码，屌到不行。其中核心实现就是takeUntil操作符，然后我就想到这个操作符的一些应用场景，例如：从网络下载序列帧zip包 > 保存本地 >解压> 转换成序列帧drawable > ui界面播放。那么当页面关闭的时候，中断这个任务链，其实调用到>解压是我想要的。于是乎我想到takeUntil这个到底放在任务链的哪部分比较合适呢？

demo走起

//创建subject，也就是监听的observable
final BehaviorSubject<String> subject = BehaviorSubject.create();
//对数据进行过滤
final Observable<String> ob =  subject.filter(new Predicate<String>() {
            @Override
            public boolean test(@NonNull String s) throws Exception {
                return s.equals("stop");
            }
        });
//创建ObservableTransformer，当监听subject成功发射数据，那么Disposable  源upstream。
ObservableTransformer<String,String> takeUntilTransformer = new ObservableTransformer<String,String>() {
            @Override
            public ObservableSource<String> apply(@NonNull Observable<String> upstream) {
                return upstream.takeUntil(ob);
            }
        };
//开始发射数据进行测试，其中compose(takeUntilTransformer)会放在2个doNext的前中后3个位置，还有subscribe之前4种情况，然后看log
Observable.just("data").compose(takeUntilTransformer).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG,"第一个doOnNext---睡眠");
                  Thread.sleep(5000);
                Log.d(TAG,"第一个doOnNext---醒来");

            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG,"第二个doOnNext---");

            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }
            @Override
            public void onNext(@NonNull String s) {
                Log.d(TAG,"subscribe---onNext---");
            }
            @Override
            public void onError(@NonNull Throwable e) {

            }
            @Override
            public void onComplete() {
                Log.d(TAG,"subscribe---onComplete---");

            }
        });
        //2秒后发送终止
        Observable.timer(2, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                Log.d(TAG,"发射-stop");
                subject.onNext("stop");
              
            }
        });

第一种情况compose(takeUntilTransformer)在发射数据之后

第一个doOnNext---睡眠
2秒后
发射-stop
3秒后
第一个doOnNext---醒来
第二个doOnNext---
subscribe---onNext---
subscribe---onComplete---

第二种情况compose(takeUntilTransformer)放在2个doNext()中间

第一个doOnNext---睡眠
2秒后
发射-stop
subscribe---onComplete---
3秒后
第一个doOnNext---醒来

第三种情况compose(takeUntilTransformer)放在2个doNext()之后

第一个doOnNext---睡眠
2秒后
发射-stop
subscribe---onComplete---
3秒后
第一个doOnNext---醒来
第二个doOnNext---

第四种情况compose(takeUntilTransformer)放在subscribe之前

第一个doOnNext---睡眠
2秒后
发射-stop
subscribe---onComplete---

好吧，4种情况打印了4种不同的结果。先看下takeUntil的实现吧

//老套路，又是返回一个Observable的实现类
    public final <U> Observable<T> takeUntil(ObservableSource<U> other) {
        ObjectHelper.requireNonNull(other, "other is null");
        return RxJavaPlugins.onAssembly(new ObservableTakeUntil<T, U>(this, other));
    }

ObservableTakeUntil

public final class ObservableTakeUntil<T, U> extends AbstractObservableWithUpstream<T, T> {
    final ObservableSource<? extends U> other;
    public ObservableTakeUntil(ObservableSource<T> source, ObservableSource<? extends U> other) {
        super(source);
        this.other = other;
    }
    @Override
    public void subscribeActual(Observer<? super T> child) {
        //当下游订阅开始后，到这里child变成了SerializedObserver ，它是用来处理并发的，最后说
        final SerializedObserver<T> serial = new SerializedObserver<T>(child);
        //存放上游的Disposable
        final ArrayCompositeDisposable frc = new ArrayCompositeDisposable(2);
        
        final TakeUntilObserver<T> tus = new TakeUntilObserver<T>(serial, frc);
        //onSubscribe在订阅处回调，不受线程影响，使各个部分对parent Disposable做处理
        child.onSubscribe(frc);
        //other是监听的Observable即subject,订阅了TakeUntil
        other.subscribe(new TakeUntil(frc, serial));
        //source为源Observable，继续订阅TakeUntilObserver
        source.subscribe(tus);
    }
    static final class TakeUntilObserver<T> extends AtomicBoolean implements Observer<T> {
        private static final long serialVersionUID = 3451719290311127173L;
        final Observer<? super T> actual;
        final ArrayCompositeDisposable frc;
        Disposable s;
        TakeUntilObserver(Observer<? super T> actual, ArrayCompositeDisposable frc) {
            this.actual = actual;
            this.frc = frc;
        }
        @Override
        public void onSubscribe(Disposable s) {
            if (DisposableHelper.validate(this.s, s)) {
                this.s = s;
                frc.setResource(0, s);
            }
        }
        @Override
        public void onNext(T t) {
            actual.onNext(t);
        }
        @Override
        public void onError(Throwable t) {
            frc.dispose();
            actual.onError(t);
        }
        @Override
        public void onComplete() {
            frc.dispose();
            actual.onComplete();
        }
    }
    final class TakeUntil implements Observer<U> {
        private final ArrayCompositeDisposable frc;
        private final SerializedObserver<T> serial;

        TakeUntil(ArrayCompositeDisposable frc, SerializedObserver<T> serial) {
            this.frc = frc;
            this.serial = serial;
        }
        @Override
        public void onSubscribe(Disposable s) {
            frc.setResource(1, s);
        }
        @Override
        public void onNext(U t) {
            frc.dispose();
            //当监听的observable成功发射数据后。立刻调用下游observer的onComplete
            //同时调用上游的dispose()方法
            serial.onComplete();
        }
        @Override
        public void onError(Throwable t) {
            frc.dispose();
            serial.onError(t);
        }
        @Override
        public void onComplete() {
            frc.dispose();
            serial.onComplete();
        }
    }
}

看到这里二三种情况就应该很好理解了
当主线程发射了stop，那么立刻调用的下游的observer的onComplete，所以立刻打印了subscribe---onComplete---，二三种情况的唯一区别的是。第二种情况没有打印第二个doOnNext，这是因为第二种情况在第一个doOnNext阻塞的时候，other发射了stop，立刻调用了下游的serial的onComplete，也就是第二个doOnNext的onComplete，在第二个doOnNext的onComplete调用后done=true是不能再接收到onNext的，所以第二个doOnNext不会被打印，那么第三种情况打印了第二个doOnNext，因为TakeUntil只作用于下游，所以第二个doOnNext不受影响。

 @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            try {
                onNext.accept(t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                s.dispose();
                onError(e);
                return;
            }
            actual.onNext(t);
        }
 @Override
        public void onComplete() {
            if (done) {
                return;
            }
            try {
                onComplete.run();
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                onError(e);
                return;
            }
            done = true;
            actual.onComplete();
            try {
                onAfterTerminate.run();
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                RxJavaPlugins.onError(e);
            }
        }

然后是第四种情况，《第一个doOnNext---醒来》的log都没有打印。证明在中断源Observable的同时将子线程干掉了。之前看过subscribeOn得实现，在撸一遍吧

 public final Observable<T> subscribeOn(Scheduler scheduler) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
    }

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;
    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }
    @Override
    public void subscribeActual(final Observer<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
        s.onSubscribe(parent);
        //看这个就明白喽。将Disposable设置成scheduler.scheduleDirect()。
        //SubscribeTask所在的线程就是scheduler指定的线程了
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }
    static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
        private static final long serialVersionUID = 8094547886072529208L;
        final Observer<? super T> actual;
        final AtomicReference<Disposable> s;
        SubscribeOnObserver(Observer<? super T> actual) {
            this.actual = actual;
            this.s = new AtomicReference<Disposable>();
        }
        @Override
        public void onSubscribe(Disposable s) {
            DisposableHelper.setOnce(this.s, s);
        }
        @Override
        public void onNext(T t) {
            actual.onNext(t);
        }
        @Override
        public void onError(Throwable t) {
            actual.onError(t);
        }
        @Override
        public void onComplete() {
            actual.onComplete();
        }
        @Override
        public void dispose() {
            DisposableHelper.dispose(s);
            DisposableHelper.dispose(this);
        }
        @Override
        public boolean isDisposed() {
            return DisposableHelper.isDisposed(get());
        }
        void setDisposable(Disposable d) {
            DisposableHelper.setOnce(this, d);
        }
    }
    final class SubscribeTask implements Runnable {
        private final SubscribeOnObserver<T> parent;

        SubscribeTask(SubscribeOnObserver<T> parent) {
            this.parent = parent;
        }
        @Override
        public void run() {
            source.subscribe(parent);
        }
    }
}

看下scheduleDirect方法

    @NonNull
    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        final Worker w = createWorker();
        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        DisposeTask task = new DisposeTask(decoratedRun, w);
        w.schedule(task, delay, unit);
        return task;
    }
 static final class DisposeTask implements Runnable, Disposable {
        final Runnable decoratedRun;
        final Worker w;
        Thread runner;
        DisposeTask(Runnable decoratedRun, Worker w) {
            this.decoratedRun = decoratedRun;
            this.w = w;
        }
        @Override
        public void run() {
            runner = Thread.currentThread();
            try {
                decoratedRun.run();
            } finally {
                dispose();
                runner = null;
            }
        }
        @Override
        public void dispose() {
            //OK ,NewThreadWorker里面有个运行子线程的线程池。调用的就线程池的shutdown()
            if (runner == Thread.currentThread() && w instanceof NewThreadWorker) {
                ((NewThreadWorker)w).shutdown();
            } else {
                w.dispose();
            }
        }
        @Override
        public boolean isDisposed() {
            return w.isDisposed();
        }
    }

当compose(takeUntilTransformer)放在subscribeOn之后时，在subscribe的时候由于subscribeActual逆向调用，所以child.onSubscribe(Dispose dispose)是调用TakeUntilObserver的

  @Override
   public void onSubscribe(Disposable s) {
        if (DisposableHelper.validate(this.s, s)) {
                this.s = s;
                frc.setResource(0, s);
            }
        }

frc就间接有了SubscribeOnObserver的引用.

还差第一种情况，第一种情况看上去是和第一个doOnNext一起阻塞了，其实不是哦，主要是因为SerializedObserver这个类，它用来处理并发，这个类的描述是这样的

When multiple threads are emitting and/or notifying they will be serialized by:
Allowing only one thread at a time to emit
Adding notifications to a queue if another thread is already emitting
Not holding any locks or blocking any threads while emitting

一次只会允许一个线程进行发送事物
如果其他线程已经准备就绪，会通知给队列
在发送事物中，不会持有任何锁和阻塞任何线程

如何处理并发

@Override
    public void onNext(@NonNull T t) {
        if (done) {
            return;
        }
        if (t == null) {
            s.dispose();
            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
            return;
        }
        synchronized (this) {
            if (done) {
                return;
            }
            if (emitting) {
                AppendOnlyLinkedArrayList<Object> q = queue;
                if (q == null) {
                    q = new AppendOnlyLinkedArrayList<Object>(QUEUE_LINK_SIZE);
                    queue = q;
                }
                q.add(NotificationLite.next(t));
                return;
            }
            emitting = true;
        }
        actual.onNext(t);
        emitLoop();
    }
    @Override
    public void onComplete() {
        if (done) {
            return;
        }
        synchronized (this) {
            if (done) {
                return;
            }
            if (emitting) {
                AppendOnlyLinkedArrayList<Object> q = queue;
                if (q == null) {
                    q = new AppendOnlyLinkedArrayList<Object>(QUEUE_LINK_SIZE);
                    queue = q;
                }
                q.add(NotificationLite.complete());
                return;
            }
            done = true;
            emitting = true;
        }
        actual.onComplete();
        // no need to loop because this onComplete is the last event
    }
    void emitLoop() {
        for (;;) {
            AppendOnlyLinkedArrayList<Object> q;
            synchronized (this) {
                q = queue;
                if (q == null) {
                    emitting = false;
                    return;
                }
                queue = null;
            }
            if (q.accept(actual)) {
                return;
            }
        }
    }

onError和onComplete同理就不贴了，有一个emitting 变量，如果一个线程调用了3个方法的某一个那么emitting 为true,当其他线程再次调用的时候那么这个操作会被加入链表数组里，当 actual.onNext(t);执行完毕会调用emitLoop();检查链表数组里是否保存了其他线程的操作，如果操作是onNext会一直调用，如果调用了onError或者onComplete那么终止。

    public <U> boolean accept(Observer<? super U> observer) {
        Object[] a = head;
        final int c = capacity;
        while (a != null) {
            for (int i = 0; i < c; i++) {
                Object o = a[i];
                if (o == null) {
                    break;
                }
                if (NotificationLite.acceptFull(o, observer)) {
                    return true;
                }
            }
            a = (Object[])a[c];
        }
        return false;
    }

NotificationLite

    public static <T> boolean acceptFull(Object o, Observer<? super T> s) {
        if (o == COMPLETE) {
            s.onComplete();
            return true;
        } else
        if (o instanceof ErrorNotification) {
            s.onError(((ErrorNotification)o).e);
            return true;
        } else
        if (o instanceof DisposableNotification) {
            s.onSubscribe(((DisposableNotification)o).d);
            return false;
        }
        s.onNext((T)o);
        return false;
    }

那么第一种情况就是主线程调用的onComplete被放到了链表数组里，待子线程的doOnNext执行完毕，然后调用了onComplete。

情况分析完毕，那么takeUntil放的位置根据不同需求进行选择了。

如有分析不对的地方请指出，Thanks~

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 202,980评论 5赞 476
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,178评论 2赞 380
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 149,868评论 0赞 336
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,498评论 1赞 273
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,492评论 5赞 364
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,521评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,910评论 3赞 395
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,569评论 0赞 256
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,793评论 1赞 296
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,559评论 2赞 319
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,639评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,342评论 4赞 318
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,931评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,904评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,144评论 1赞 259
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,833评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,350评论 2赞 342

推荐阅读更多精彩内容

  • Rxjava使用详解，强烈推荐
    最近项目里面有用到Rxjava框架，感觉很强大的巨作，所以在网上搜了很多相关文章，发现一片文章很不错，今天把这篇文...
    Scus阅读 6,852评论 2赞 50
  • RxJava操作符（一）
    注：只包含标准包中的操作符，用于个人学习及备忘参考博客：http://blog.csdn.net/maplejaw...
    小白要超神阅读 2,182评论 2赞 8
  • <<转>>给 Android 开发者的 RxJava 详解
    前言我从去年开始使用 RxJava ，到现在一年多了。今年加入了 Flipboard 后，看到 Flipboard...
    占导zqq阅读 9,158评论 6赞 151
  • RxJava 2.x 学习（转载）
    转载自：https://xiaobailong24.me/2017/03/18/Android-RxJava2.x...
    Young1657阅读 2,007评论 1赞 9
  • 麻烦的狠
    不知为何最不想去的地方就是医院，但为了延续生命和那求生的欲望，又不得不去医院。 有时医院就像救命的稻草，他会挽留你...
    宝鱼阅读 215评论 1赞 1