RxJava源码解析05-线程切换

爱被打了一巴掌 2023-10-17 15:46 74阅读 0赞

### RxJava源码解析05-线程切换 ###

这篇文章我们来分析一下RxJava线程的切换。

Observable.just("123")
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.newThread())
        .subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
    
            }
    
            @Override
            public void onError(Throwable e) {
    
            }
    
            @Override
            public void onNext(String s) {
    
            }
        });

#### subscribeOn ####

observaOn作用于该操作符之后直到出现新的observeOn操作符，在Observer.onNext、onCompleted、onError的时候进行切换。subscribeOn可以进行多次切换。

![1240][]

#### subscribeOn ####

subscribeOn作用于该操作符之前的Observable的**创建操作符**以及doOnSubscribe操作符，subscribeOn有多个时，只有第一个有效(除doSubscriber外，doSubscribe会在OnSubscribe.call前调用)

在Onsubscribe.call()时进行切换

#### subscribeOn源码解析 ####

public final Observable<T> subscribeOn(Scheduler scheduler, boolean requestOn) {
        if (this instanceof ScalarSynchronousObservable) {
            return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
        }
        return unsafeCreate(new OperatorSubscribeOn<T>(this, scheduler, requestOn));
    }

来看OperatorSubscribeOn

@Override
    public void call(final Subscriber<? super T> subscriber) {
        final Worker inner = scheduler.createWorker();
    
        SubscribeOnSubscriber<T> parent = new SubscribeOnSubscriber<T>(subscriber, requestOn, inner, source);
        subscriber.add(parent);
        subscriber.add(inner);
    
        inner.schedule(parent);
    }

这里会调用`scheduler.createWorker();`创建Worker，并调用schedule。而这个scheduler是构造方法传参进来的，这是调用subscribeOn()传入的参数。

来看下Schedulers.newThread()，跟踪源码会发现，是NewThreadScheduler类

public final class NewThreadScheduler extends Scheduler {
        private final ThreadFactory threadFactory;
    
        public NewThreadScheduler(ThreadFactory threadFactory) {
            this.threadFactory = threadFactory;
        }
    
        @Override
        public Worker createWorker() {
            return new NewThreadWorker(threadFactory);
        }
    }

主要来看NewThreadWorker

public NewThreadWorker(ThreadFactory threadFactory) {
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory);
    
        boolean cancelSupported = tryEnableCancelPolicy(exec);
        if (!cancelSupported && exec instanceof ScheduledThreadPoolExecutor) {
            registerExecutor((ScheduledThreadPoolExecutor)exec);
        }
        executor = exec;
    }

可以看到这里创建了线程池，再来看schedule方法

@Override
    public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
        if (isUnsubscribed) {
            return Subscriptions.unsubscribed();
        }
        return scheduleActual(action, delayTime, unit);
    }  
    
    public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) {
        Action0 decoratedAction = RxJavaHooks.onScheduledAction(action);
        ScheduledAction run = new ScheduledAction(decoratedAction);
        Future<?> f;
        if (delayTime <= 0) {
            f = executor.submit(run);
        } else {
            f = executor.schedule(run, delayTime, unit);
        }
        run.add(f);
    
        return run;
    }

可以看到这里调用了`executor.submit()`，action是之前OperatorSubscribeOn.schedule传入的parent对象，这是个Action0的子类，最终会调用parent.call。parent是SubscribeOnSubscriber

@Override
    public void call() {
        //...
        src.unsafeSubscribe(this); //src是subscribe
    }

从而onNext、onCompleted、onError的时候，就已经在切换后的线程中执行了，从而实现了线程的切换。

@Override
    public void onNext(T t) {
        actual.onNext(t);
    }
    
    @Override
    public void onError(Throwable e) {
        try {
            actual.onError(e);
        } finally {
            worker.unsubscribe();
        }
    }
    
    @Override
    public void onCompleted() {
        try {
            actual.onCompleted();
        } finally {
            worker.unsubscribe();
        }
    }

#### observeOn 源码分析 ####

经过一些方法，会调用Observer\#observeOn()

public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
        //...
        return lift(new OperatorObserveOn<T>(scheduler, delayError, bufferSize));
    }

lift方法会将传入的T转换为R，这里的OperatorObserveOn继承自Operator

#### 总结 ####

RxJava的线程切换其实就是根据Schedulers生成的Scheduler，最终调用相应的线程池来进行线程切换。

如果是或AndroidSchedulers.mainThread，则会使用Handler来切换到主线程。

class LooperScheduler extends Scheduler {
        //...       
    
        @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;
        }
    
        //...
    }

[1240]: http://upload-images.jianshu.io/upload_images/1724103-95f7846f67bad66f.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240