RxJava2 源碼淺析

RxJava2 源碼淺析

ReactiveX

歷史：
ReactiveX是Reactive Extensions的縮寫，一般簡寫為Rx，最初是LINQ的一個擴(kuò)展，由微軟的架構(gòu)師Erik Meijer領(lǐng)導(dǎo)的團(tuán)隊開發(fā)，在2012年11月開源，Rx是一個編程模型，目標(biāo)是提供一致的編程接口，幫助開發(fā)者更方便的處理異步數(shù)據(jù)流，Rx庫支持.NET、JavaScript和C++，Rx近幾年越來越流行了，現(xiàn)在已經(jīng)支持幾乎全部的流行編程語言了，Rx的大部分語言庫由ReactiveX這個組織負(fù)責(zé)維護(hù)，比較流行的有RxJava/RxJS/Rx.NET，社區(qū)網(wǎng)站是 reactivex.io

定義：
ReactiveX.io給的定義是，Rx是一個使用可觀察數(shù)據(jù)流進(jìn)行異步編程的編程接口，ReactiveX結(jié)合了觀察者模式、迭代器模式和函數(shù)式編程的精華。
ReactiveX不僅僅是一個編程接口，它是一種編程思想的突破，它影響了許多其它的程序庫和框架以及編程語言。

filter操作符

這就是數(shù)據(jù)流？

RxJava2定義

a library for composing asynchronous and event-based programs by using observable sequences.(一個在 Java VM 上使用可觀測的序列來組成異步的、基于事件的程序的庫）

大致流程

上代碼：

Observable.
            create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    emitter.onNext("test First");
                    Log.e("TAG", "subScribe test First");
                    emitter.onNext("test Second");
                    Log.e("TAG", "subScribe test Second");
                    emitter.onComplete();
                    Log.e("TAG", "subScribe onComplete");
                }
            }).
            subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.e("TAG", "onSubscribe");
                    mDisposable = d;
                }

                @Override
                public void onNext(String s) {
                    Log.e("TAG", "onNext");
                    Log.e("TAG", s);
                    if (s.equals("test First")) {
                        mDisposable.dispose();
                    }
                }

                @Override
                public void onError(Throwable e) {
                    Log.e("TAG", "onError");
                }

                @Override
                public void onComplete() {
                    Log.e("TAG", "onComplete");
                }
            });

剛開始學(xué)習(xí)RxJava 時，這段代碼給我最直觀的感受就是，這不就是自己調(diào)用自己嗎。ObservableEmitter<String> emitter 這個就是下面的subscribe(new Observer<String>()）。對吧，我覺得大家應(yīng)該都是這樣的感受吧...

追蹤一下源碼：點擊create()方法進(jìn)去看一下：

@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}


public static <T> Observable<T> onAssembly(Observable<T> source) {
    Function<Observable, Observable> f = onObservableAssembly;
    if (f != null) {
        return apply(f, source);
    }
    return source;
}
看了代碼onObservableAssembly為null,所以create方法之后這個對象被包裝成new ObservableCreate<T>(source)，source是外面?zhèn)鬟M(jìn)來的。

關(guān)鍵字：io.reactivex.internal.operators.observable.ObservableCreate

繼續(xù)看下一個操作符：subscribe()，點進(jìn)去看一下

@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);//observer原樣返回，沒改動

        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");

        subscribeActual(observer);//關(guān)鍵點
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Subscription already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}
// RxJavaPlugins.onSubscribe

  public static <T> Observer<? super T> onSubscribe(Observable<T> source, Observer<? super T> observer) {
    BiFunction<Observable, Observer, Observer> f = onObservableSubscribe;   //f 為null
    if (f != null) {
        return apply(f, source, observer);
    }
    return observer;
}

上面說過了經(jīng)過create()方法或這個對象已經(jīng)是ObservableCreate了，那么最終會調(diào)用的就是subscribeActual(observer) 看一下ObservableCreate這個類的代碼：

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

    try {
        source.subscribe(parent);
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        parent.onError(ex);
    }
}


static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
CreateEmitter(Observer<? super T> observer) {
        this.observer = observer;
    }

    @Override
    public void onNext(T t) {
        if (t == null) {
            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
            return;
        }
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }
}

看一下，這里會把observer包裝成一個CreateEmitter對象，然后source是Observable.create(new ObservableOnSubscribe<String>())傳進(jìn)來的ObservableOnSubscribe對象。然后會調(diào)用observer.onSubscribe(parent);source.subscribe(parent);終于清晰了...可以回答上面的問題了，其實Observer和ObservableEmitter可以看成是一個對象，只是對observer做了個包裝...

Scheduler 線程變換(subscribeOn 和 observeOn)

說到線程變換即線程間通信，因為我是學(xué)Android，所以第一印象就是Handler，然后就是Future?？戳嗽创a后發(fā)現(xiàn)RxJava用的是Future，ScheduledExecutorService，Runnable，二AndroidScheduler就是用Handler的，因為需要切換到Android中的UI線程。

subscribeOn(Schedulers.newThread())

點進(jìn)去看一下：

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);

    parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
        @Override
        public void run() {
            source.subscribe(parent);
        }
    }));
}
}

傳進(jìn)來的是Schedulers.newThread()，點擊Schedulers.newThread() 點進(jìn)去
發(fā)現(xiàn)最終返回的是NewThreadScheduler

關(guān)鍵字：io.reactivex.internal.schedulers.NewThreadScheduler
io.reactivex.internal.schedulers.NewThreadWorker(真正做線程調(diào)度的類)

發(fā)現(xiàn)有scheduler.scheduleDirect(new Runnable()）點擊進(jìn)去，最終調(diào)用
public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
final Worker w = createWorker();

    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    w.schedule(new Runnable() {
        @Override
        public void run() {
            try {
                decoratedRun.run();
            } finally {
                w.dispose();
            }
        }
    }, delay, unit);

    return w;
}
最終還是w.schedule(new Runnable()),w就是NewThreadWorker，找到這個類看一下schedule方法,最終會調(diào)用：


public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, TimeUnit unit, DisposableContainer parent) {
    Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

    if (parent != null) {
        if (!parent.add(sr)) {
            return sr;
        }
    }

    Future<?> f;
    try {
        if (delayTime <= 0) {
            f = executor.submit((Callable<Object>)sr);
        } else {
            f = executor.schedule((Callable<Object>)sr, delayTime, unit);
        }
        sr.setFuture(f);
    } catch (RejectedExecutionException ex) {
        parent.remove(sr);
        RxJavaPlugins.onError(ex);
    }

    return sr;
}

看到future和executor了，這里就是線程切換

observeOn(AndroidSchedulers.mainThread())
ObservableObserveOn 最終調(diào)用的是HandlerScheduler和HandlerWorker

HandlerWorker:

@Override
    public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
        if (run == null) throw new NullPointerException("run == null");
        if (delay < 0) throw new IllegalArgumentException("delay < 0: " + delay);
        if (unit == null) throw new NullPointerException("unit == null");

        if (disposed) {
            return Disposables.disposed();
        }

        run = RxJavaPlugins.onSchedule(run);

        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

        Message message = Message.obtain(handler, scheduled);
        message.obj = this; // Used as token for batch disposal of this worker's runnables.

        handler.sendMessageDelayed(message, unit.toMillis(delay));

        // Re-check disposed state for removing in case we were racing a call to dispose().
        if (disposed) {
            handler.removeCallbacks(scheduled);
            return Disposables.disposed();
        }

        return scheduled;
    }

發(fā)現(xiàn)是用Handler來做線程切換，Handler管理的Looper是Looper.getMainLooper()，所以把消息發(fā)送到了主線程。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容