博客

RxJava源码解析

  • Post author:
  • Post category:java


1、基本原理

RxJava基本原理分析


https://www.jianshu.com/p/88aacbed8aa5

示例 

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                KLog.e(TAG, "1");
                emitter.onNext("1");
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                KLog.e(TAG, "2");
            }
        }).map(new Function<String, Boolean>() {
            @Override
            public Boolean apply(String resultTMPResult) throws Exception {
                KLog.e(TAG, "4");
                return true;
            }
        }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                KLog.e(TAG, "5");
            }
        });

从上向下执行，逐个初始化各个Observable，保存

上级关系属性Source

。

ObservableCreate

ObservableDoOnEach

ObservableMap

以上均继承Observable抽象类，是Observable的子类。



各Observable的具体子类的属性中，


Source代表上一级的被观察者。


对象本身 是相对上一级的观察者、和相对下一级的被观察者。


t/actual/downstream代表下一级的观察者
 

public abstract class Observable<T> implements ObservableSource<T> {
   //Observable实现ObservableSource接口
    @Override
    public final void subscribe(Observer<? super T> observer) {
        //调用每个Observable子类的subscribeActual实现方法
        subscribeActual(observer);
    }
}

//
public interface ObservableSource<T> {
    //上一级订阅本级
    void subscribe(@NonNull Observer<? super T> observer);
}

最后执行到subscribe()方法时，会最终调用Observable.subscribeActual()抽象方法。

触发依次向上执行。

ObservableMap的subscribeActual()方法。



source.subscribe(new MapObserver<T, U>(t, function));



上一级Observable订阅包含下一级Observer对象的本级Observer

 

    @Override
    public void subscribeActual(Observer<? super U> t) {
        KLog.e(TAG,"subscribeActual");
        source.subscribe(new MapObserver<T, U>(t, function));
    }

ObservableDoOnEach的subscribeActual()方法 

    @Override
    public void subscribeActual(Observer<? super T> t) {
        KLog.e(TAG,"subscribeActual");
        source.subscribe(new DoOnEachObserver<T>(t, onNext, onError, onComplete, onAfterTerminate));
    }

ObservableCreate的subscribeActual()方法 

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        KLog.e(TAG,"subscribeActual");
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        KLog.e(TAG,"subscribeActual onSubscribe");
        observer.onSubscribe(parent);

        try {
            KLog.e(TAG,"subscribeActual subscribe");
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

在

ObservableCreate

中的subscribeActual方法，会首先调用observer.onSubscribe(parent)，依次向下触发已订阅的回调。

然后调用source.subscribe(parent)方法，触发create原始接口回调的业务内容，结束后调用emitter.onNext()，触发下一级onNext()方法。

以ObservableDoOnEach的onNext()为例，上一级触发onNext()后，先执行本级的onNext()，再执行下一级的onNext()。 

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            try {
                KLog.e(TAG,"onNext1");
                onNext.accept(t);//本机业务
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                upstream.dispose();
                onError(e);
                return;
            }

            KLog.e(TAG,"onNext2");
            downstream.onNext(t);//触发下一级
        }

初始化：由上到下。

订阅：从下到上。

业务执行：从上到下。

2、切换线程

详解 RxJava2 的线程切换原理


https://www.jianshu.com/p/a9ebf730cd08

示例 

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                KLog.e(TAG, "1=" + Thread.currentThread());
                emitter.onNext("aaa");
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                KLog.e(TAG, "2=" + Thread.currentThread());
            }
        }).subscribeOn(Schedulers.io()).map(new Function<String, Boolean>() {
            @Override
            public Boolean apply(String resultTMPResult) throws Exception {
                KLog.e(TAG, "3=" + Thread.currentThread());
                return true;
            }
        }).doOnNext(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean s) throws Exception {
                KLog.e(TAG, "4=" + Thread.currentThread());
            }
            //切换线程
        }).observeOn(Schedulers.newThread()).doOnNext(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                KLog.e(TAG, "5=" + Thread.currentThread());
            }
        }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                KLog.e(TAG, "6=" + Thread.currentThread());
            }
        });

结果： 

1=Thread[RxCachedThreadScheduler-1,5,main]
2=Thread[RxCachedThreadScheduler-1,5,main]
3=Thread[RxCachedThreadScheduler-1,5,main]
4=Thread[RxCachedThreadScheduler-1,5,main]
5=Thread[RxNewThreadScheduler-1,5,main]
6=Thread[RxNewThreadScheduler-1,5,main]

订阅切换线程 ：

subscribeOn(Schedulers.io()) 生成ObservableSubscribeOn类 

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {   
    @Override
    public void subscribeActual(final Observer<? super T> observer) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);

        //触发订阅回调
        observer.onSubscribe(parent);

        //切换线程
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }
}

static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {

        @Override
        public void onSubscribe(Disposable d) {
            //订阅回调，Disposable d实际是上一级对象的Disposable。用于取消订阅使用
            DisposableHelper.setOnce(this.upstream, d);
        }

        @Override
        public void onNext(T t) {
            //上一级已切换线程，这里不切换线程，直接执行下一级onNext()
            downstream.onNext(t);
        }

}


public abstract class Scheduler {

    @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;
    }
}

    final class SubscribeTask implements Runnable {
        private final SubscribeOnObserver<T> parent;

        SubscribeTask(SubscribeOnObserver<T> parent) {
            this.parent = parent;
        }

        @Override
        public void run() {
            //在新线程里订阅
            source.subscribe(parent);
        }
    }

观察切换线程：

observeOn(Schedulers.newThread())生成ObservableObserveOn类 

public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        //新建线程
        Scheduler.Worker w = scheduler.createWorker();

        //订阅不切换线程，观察对象中传入新建线程
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}

  static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
    implements Observer<T>, Runnable {


        @Override
        public void onNext(T t) {
            //切换线程执行
            schedule();
        }

        void schedule() {
            if (getAndIncrement() == 0) {
                //切换线程执行下一级onNext()
                worker.schedule(this);
            }
        }

  }


Tags: