rxJava学习笔记之observeOn

╰+哭是因爲堅強的太久メ 2022-07-14 04:08 157阅读 0赞

一段最简单的代码

longOpeSubscription = Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
                    printLog("onStart in OnSubscribe");
                    subscriber.onStart();
                    int N = data.length;
                    for (int i = 0; i < N; i++) {
                        dosomethingBlockThread();
                        printLog("onNext" + data[i] + " in OnSubscribe");
                        subscriber.onNext(data[i]);
    
                    }
                    printLog("OnCompleted in OnSubscribe");
                    subscriber.onCompleted();
                }
            })
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Subscriber<String>() {
                        @Override
                        public void onCompleted() {
                            printLog("OnCompleted in Subscriber");
    //                        mProgressOperationRunning.setVisibility(View.INVISIBLE);
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            printLog("onError in Subscriber");
    //                        mProgressOperationRunning.setVisibility(View.INVISIBLE);
    
                        }
    
                        @Override
                        public void onNext(String s) {
                            printLog("onNext " + s + " in Subscriber");
    
                        }
                    });

和前两篇同样的分析，observeOn最终是通过lift变换来实现的，这里具体的对象是onSubscribeLift，该对象保存第一次的onSubscribe和operator，operator在这里具体是OperatorObserveOn对象。subscribe会调用observeOn中生成的obervable中的onSubscribe对象中call方法也就是onSubscribeLift的call方法。

@Override
        public void call(Subscriber<? super R> o) {
            try {
                Subscriber<? super T> st = hook.onLift(operator).call(o);
                try {
                    // new Subscriber created and being subscribed with so 'onStart' it
                    st.onStart();
                    parent.call(st);
                } catch (Throwable e) {
                    // localized capture of errors rather than it skipping all operators 
                    // and ending up in the try/catch of the subscribe method which then
                    // prevents onErrorResumeNext and other similar approaches to error handling
                    Exceptions.throwIfFatal(e);
                    st.onError(e);
                }
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                // if the lift function failed all we can do is pass the error to the final Subscriber
                // as we don't have the operator available to us
                o.onError(e);
            }
        }

这里首先执行operator的call方法，这里的o也就是下面的child就是最终的subscriber，进入OperatorObserveOn中call中

@Override
        public Subscriber<? super T> call(Subscriber<? super T> child) {
            if (scheduler instanceof ImmediateScheduler) {
                // avoid overhead, execute directly
                return child;
            } else if (scheduler instanceof TrampolineScheduler) {
                // avoid overhead, execute directly
                return child;
            } else {
                ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError, bufferSize);
                parent.init();
                return parent;
            }
        }

这里最终会返回一个parent对象，也就是ObserveOnSubscriber，其中的scheduler就是我们写的ndroidSchedulers.mainThread()。分析到这里我们回到onSubscribeLift的call方法中的 parent.call(st)这行代码，这里的parent就是create中传入的onSubscribe，这里我们再次调用了st的onNext方法，我们再次跳转到OperatorObserveOn.onNext中：

@Override
            public void onNext(final T t) {
                if (isUnsubscribed() || finished) {
                    return;
                }
                if (!queue.offer(on.next(t))) {
                    onError(new MissingBackpressureException());
                    return;
                }
                schedule();
            }

注意这里if (!queue.offer(on.next(t)))，会将值传入queue中。的最终会进入 recursiveScheduler.schedule(this);这个recursiveScheduler又是什么呢，看源码得知这个是由AndroidSchedulers.mainThread().createWorker()产生的，我们再次寻找，最终recursiveScheduler定位到一个HandlerWorker对象，我们看看HandlerWorker的schedule方法

@Override
            public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
                if (unsubscribed) {
                    return Subscriptions.unsubscribed();
                }
    
                action = hook.onSchedule(action);
    
                ScheduledAction scheduledAction = new ScheduledAction(action, handler);
    
                Message message = Message.obtain(handler, scheduledAction);
                message.obj = this; // Used as token for unsubscription operation.
    
                handler.sendMessageDelayed(message, unit.toMillis(delayTime));
    
                if (unsubscribed) {
                    handler.removeCallbacks(scheduledAction);
                    return Subscriptions.unsubscribed();
                }
    
                return scheduledAction;
            }

有没有很熟悉，这个handler自然是使用主线程looper的handler了，scheduledAction的run最终会调用action的call，这是一个无参方法，定位到OperatorObserveOn的call，调用最终的subscripter的onNext方法。oberservOn分析完毕