0章 RxJava2课程目录介绍

时间 2019-11-16

原文原文链接

本篇文章已受权微信公众号 YYGeeker 独家发布转载请标明出处bash

10. RxJava源码分析

RxJava源码分析最主要的点在于多线程

RxJava是如何从事件流的发送方发送到事件流的接收方的
RxJava是如何对操做符进行封装和操做的
RxJava是如何随意切换线程的

在分析的过程当中，部分源码分析咱们会经过手写RxJava的部分代码进行分析，固然也会结合实际RxJava的代码进行分析。其中，手写RxJava的缘由是为了简化源代码，让读者方便阅读到主要代码，更快的看懂RxJava的实现思路。在阅读源码以前，咱们须要对RxJava的大致概念进行简单的梳理app

发射器：Emitter，发射数据的对象
被观察者：Observable，被观察的对象
观察者：Observer，观察的对象
被观察者被订阅时：ObservableOnSubscribe，被订阅时的回调，同时建立出发射器
释放者：Disposable，释放RxJava的对象

RxJava的分析三步骤composer

建立：被观察者建立的过程
订阅：被观察者订阅观察者的过程
发射：发射器发射的过程

RxJava原理图解框架

第一排表示各个对象的建立关系，A->B->C->D
第二排表示各个对象的订阅关系，D->C->B->A
第三排表示各个对象的发射关系，A->B->C->D

10.1 RxJava的事件发射原理

知识点：ide

理解发射数据的过程
理解接收数据的过程

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(Emitter<String> emitter) {
        emitter.onNext("Hello RxJava");
        emitter.onError();
        emitter.onNext("Hello RxJava");
    }
}).subscribe(new Observabler<String>() {
    @Override
    public void onSubscribe() {
        
    }

    @Override
    public void onNext(String string) {
        System.out.println("onNext=" + string);
    }

    @Override
    public void onError() {
        System.out.println("onError");
    }

    @Override
    public void onComplete() {

    }
});
复制代码

输出结果：在输出onError后，就不会继续收到新的事件流，表示事件已经被释放了函数

onNext=Hello RxJava
onError
复制代码

一、定义接口工具

发射器

public interface Emitter<T> {
    void onNext(T t);
    void onError();
}
复制代码

观察者

public interface Observer<T> {
    void onSubscribe();
    void onNext(T t);
    void onError();
    void onComplete();
}
复制代码

被观察者被订阅时

public interface ObservableOnSubscribe<T> {
    void subscribe(Emitter<T> emitter);
}
复制代码

二、实现被观察者

被观察者Observable负责建立、订阅，发射由发射器负责

建立：建立的过程只是将传递进来的参数交给新的ObservableCreate进行管理
订阅：订阅的过程只是实现建立出来的ObservableCreate的subscribeActual方法

public abstract class Observable<T> {

    public static <T> ObservableCreate create(ObservableOnSubscribe<T> observableOnSubscribe) {
        return new ObservableCreate<T>(observableOnSubscribe);
    }

    public void subscribe(Observer<T> observer) {
        subscribeActual(observer);
    }

    public abstract void subscribeActual(Observer<T> observer);
}
复制代码

三、ObservableCreate

ObservableCreate继承自Observable，因为Observable.create返回当前ObservableCreate，因此在subscribe的时候，走的是这里的subscribeActual，subscribeActual中会去建立发射器，并给发射器传递进去observer

public class ObservableCreate<T> extends Observable{

    private ObservableOnSubscribe source;

    public ObservableCreate(ObservableOnSubscribe observableOnSubscribe) {
        this.source = observableOnSubscribe;
    }

    @Override
    public void subscribeActual(Observer observer) {
        //固定的三步曲分析法（我的建立，基本都是这个步骤）
        
        //一、建立发射器
        EmitterCreate<T> emitterCreate = new EmitterCreate<>(observer);
        //二、回调observer的onSubscribe
        observer.onSubscribe();
        //三、回调上一个的subscribe
        source.subscribe(emitterCreate);
    }
}
复制代码

四、EmitterCreate

传递进来的observer便是咱们最开始订阅时候new出来的，此时发射数据，就会去调用Observer的onNext方法，这样数据就从发射器中传递到观察者中了。DisposableHelper在后面会讲到，这里只是用做判断是否被释放的一个工具类

public class EmitterCreate<T>
        extends AtomicReference<Disposable>
        implements Emitter<T>, Disposable {

    private Observer<T> observer;

    public EmitterCreate(Observer<T> observer) {
        this.observer = observer;
    }

    @Override
    public void onNext(T t) {
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }

    @Override
    public void onError() {
        if (!isDisposed()) {
            try {
                observer.onError();
            } finally {
                dispose();
            }
        }
    }

    @Override
    public void dispose() {
        DisposableHelper.dispose(this);
    }

    @Override
    public boolean isDisposed() {
        return DisposableHelper.isDisposed(get());
    }
}
复制代码

一、Observable.create

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");//判空
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));//返回自身
}
复制代码

RxJavaPlugins.onAssembly只是对传递进来的参数作判断处理，最终仍是返回ObservableCreate，有关RxJavaPlugins的东西最终都是返回自身，RxJavaPlugins后面分析会说到，这里只须要知道他是返回参数自己便可

二、Observable.subscribe

public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");//判空
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);//返回自身

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

        subscribeActual(observer);//回调ObservableCreate的subscribeActual
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        ......
        throw npe;
    }
}
复制代码

observable.subscribe和咱们手写代码同样，最终调用的是ObservableCreate的subscribeActual方法

三、ObservableCreate

public final class ObservableCreate<T> extends Observable<T> {
    final ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        //一、建立发射器
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        //二、回调observer的onSubscribe
        observer.onSubscribe(parent);

        try {
            //三、回调ObservableOnSubscribe的subscribe
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

    static final class CreateEmitter<T>
    extends AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable {


        private static final long serialVersionUID = -3434801548987643227L;

        final Observer<? super T> observer;

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

        @Override
        public void onError(Throwable t) {
            if (!tryOnError(t)) {
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public boolean tryOnError(Throwable t) {
            if (t == null) {
                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
            }
            if (!isDisposed()) {
                try {
                    observer.onError(t);
                } finally {
                    dispose();
                }
                return true;
            }
            return false;
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }

        @Override
        public void setDisposable(Disposable d) {
            DisposableHelper.set(this, d);
        }

        @Override
        public void setCancellable(Cancellable c) {
            setDisposable(new CancellableDisposable(c));
        }

        @Override
        public ObservableEmitter<T> serialize() {
            return new SerializedEmitter<T>(this);
        }

        @Override
        public void dispose() {
            DisposableHelper.dispose(this);
        }

        @Override
        public boolean isDisposed() {
            return DisposableHelper.isDisposed(get());
        }
    }
}
复制代码

ObservableCreate和咱们手写代码同样，建立发射器，并在发射器中作发射数据等操做

小结

如图所示

10.2 RxJava的事件释放原理

知识点：

理解释放事件的原理

有关RxJava的释放原理是基于Observable能够返回Disposable对象，只有调用dispose()才能释放事件，经过上面的例子，咱们知道在发射器里面有isDisposed()和dispose()操做，在发射完onError事件的状况下，咱们会将事件释放，因此在finally会作释放操做，防止后面的事件再次发射

public class EmitterCreate<T>
        extends AtomicReference<Disposable>
        implements Emitter<T>, Disposable {

    private Observer<T> observer;

    public EmitterCreate(Observer<T> observer) {
        this.observer = observer;
    }

    @Override
    public void onNext(T t) {
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }

    @Override
    public void onError() {
        if (!isDisposed()) {
            try {
                observer.onError();
            } finally {
                dispose();
            }
        }
    }

    @Override
    public void dispose() {
        DisposableHelper.dispose(this);
    }

    @Override
    public boolean isDisposed() {
        return DisposableHelper.isDisposed(get());
    }
}
复制代码

@Override
public void dispose() {
    DisposableHelper.dispose(this);
}

@Override
public boolean isDisposed() {
    return DisposableHelper.isDisposed(get());
}
复制代码

能够发现事件的释放都是经过DisposableHelper去触发的，不论是手写RxJava仍是源代码，释放RxJava都是经过DisposableHelper进行释放，具体看DisposableHelper。在咱们的演示程序中，咱们经过发射onNext->onError->onNext的过程，去挖掘事件是怎么被释放掉的

public enum DisposableHelper implements Disposable {
    
    DISPOSED
    ;

    public static boolean isDisposed(Disposable d) {
        return d == DISPOSED;
    }

    public static boolean dispose(AtomicReference<Disposable> field) {
        Disposable current = field.get();//获取参数的Disposable对象
        Disposable d = DISPOSED;//声明一个已经释放的Disposable对象
        if (current != d) {//若是当前未被释放
            current = field.getAndSet(d);//则将当前的Disposable赋值成已经释放过的Disposable对象
            if (current != d) {//若是当前还未被释放
                if (current != null) {//且不为空
                    current.dispose();//则释放当前Disposable对象
                }
                return true;
            }
        }
        return false;
    }
}
复制代码

在事件释放的过程当中，EmitterCreate自己是个AtomicReference<Disposable>，代码经过get()去获取Disposable对象，其中代码会经过双层判断去作释放，防止在多线程的时候出现抢夺的状况

onNext：第一次发射数据时，get()会获取一个null对象，因此不符合d == DISPOSED
onEroor：这时候会调用dispose()去比较当前和释放过的对象，若是不等于，则将当前的对象设置为释放过的值
onNext：第二次发射数据时，get()会获取一个已经释放过的对象，这个时候符合d == DISPOSED

其实这里的操做如同设置一个Flag，但因为Disposable是对象的形式，且须要保证原子性，AtomicReference类型是个最佳选择，能保证对象的原子性

10.3 RxJava的背压原理

知识点：

理解背压实现的本质
理解背压数据项丢弃的本质

背压原理有一部分和RxJava事件发射原理类似，其背压的过程就是在不一样策略的发射器去处理当前的数据项而已。在分析背压策略的时候，咱们都知道背压是须要手动进行请求才能够将数据发射到观察者中，因此咱们会调用s.request(Long.MAX_VALUE)让观察者能接收到数据。有些人就会有疑问，为何有些人平时用背压的时候，不须要去调用request()就能接收到数据，缘由是有些背压已经在内部默认调用了s.request(Long.MAX_VALUE)，因此这里是不用多想的，是必定要调用s.request(Long.MAX_VALUE)才能收到数据的。因为不一样背压的策略的原理大同小异，主要以Drop策略去分析背压的原理

public static void drop(View view) {
    Flowable.create(new FlowableOnSubscribe<Integer>() {
        @Override
        public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
            for (int i = 0; i < 1000; i++) {
                emitter.onNext(i);
            }
        }
    }, BackpressureStrategy.DROP)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new FlowableSubscriber<Integer>() {
                @Override
                public void onSubscribe(Subscription s) {
                    s.request(Long.MAX_VALUE);
                }

                @Override
                public void onNext(Integer integer) {
                    Log.e("TAG", "onNext=" + integer);
                }

                @Override
                public void onError(Throwable t) {
                    t.printStackTrace();
                }

                @Override
                public void onComplete() {

                }
            });
}
复制代码

一、Flowable.create

public static <T> Flowable<T> create(FlowableOnSubscribe<T> source, BackpressureStrategy mode) {
    ObjectHelper.requireNonNull(source, "source is null");//判空
    ObjectHelper.requireNonNull(mode, "mode is null");//判空
    return RxJavaPlugins.onAssembly(new FlowableCreate<T>(source, mode));//返回自身
}
复制代码

Flowable.create跟咱们前面是同样的，最后仍是会交给新对象FlowableCreate去处理

二、Flowable.subscribe

public final void subscribe(FlowableSubscriber<? super T> s) {
    ObjectHelper.requireNonNull(s, "s is null");
    try {
        Subscriber<? super T> z = RxJavaPlugins.onSubscribe(this, s);

        ObjectHelper.requireNonNull(z, "Plugin returned null Subscriber");

        subscribeActual(z);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        ......
        throw npe;
    }
}
复制代码

Flowable.subscribe跟咱们前面是同样的，最终调用的是FlowableCreate的subscribeActual方法

三、FlowableCreate.subscribeActual

public final class FlowableCreate<T> extends Flowable<T> {

    final FlowableOnSubscribe<T> source;

    final BackpressureStrategy backpressure;

    public FlowableCreate(FlowableOnSubscribe<T> source, BackpressureStrategy backpressure) {
        this.source = source;
        this.backpressure = backpressure;
    }

    @Override
    public void subscribeActual(Subscriber<? super T> t) {
        //使用三步曲分析法
        
        BaseEmitter<T> emitter;

        switch (backpressure) {
        case MISSING: {
            emitter = new MissingEmitter<T>(t);
            break;
        }
        case ERROR: {
            emitter = new ErrorAsyncEmitter<T>(t);
            break;
        }
        case DROP: {
            emitter = new DropAsyncEmitter<T>(t);
            break;
        }
        case LATEST: {
            emitter = new LatestAsyncEmitter<T>(t);
            break;
        }
        default: {
            emitter = new BufferAsyncEmitter<T>(t, bufferSize());
            break;
        }
        }

        t.onSubscribe(emitter);
        try {
            source.subscribe(emitter);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            emitter.onError(ex);
        }
    }
}
复制代码

subscribeActual会根据不一样的策略生成不一样的发射器，具体的全部策略逻辑都在发射器中体现的

四、DropAsyncEmitter

static final class DropAsyncEmitter<T> extends NoOverflowBaseAsyncEmitter<T> {

    private static final long serialVersionUID = 8360058422307496563L;

    DropAsyncEmitter(Subscriber<? super T> actual) {
        super(actual);
    }

    @Override
    void onOverflow() {
        // nothing to do
    }

}
复制代码

DropAsyncEmitter其实没作什么事情，主要都在其父类中实现了，onOverflow的回调表示事件流溢出的时候的处理，很明显Drop策略就把溢出的数据项直接不作处理，意思就是抛弃掉这个数据项了

static final class ErrorAsyncEmitter<T> extends NoOverflowBaseAsyncEmitter<T> {

    private static final long serialVersionUID = 338953216916120960L;

    ErrorAsyncEmitter(Subscriber<? super T> actual) {
        super(actual);
    }

    @Override
    void onOverflow() {
        onError(new MissingBackpressureException("create: could not emit value due to lack of requests"));
    }

}
复制代码

再看看Error策略，溢出以后就会抛出溢出的异常，其余策略也相似分析，具体父类是如何处理溢出函数的呢

五、NoOverflowBaseAsyncEmitter

abstract static class NoOverflowBaseAsyncEmitter<T> extends BaseEmitter<T> {

    private static final long serialVersionUID = 4127754106204442833L;

    NoOverflowBaseAsyncEmitter(Subscriber<? super T> actual) {
        super(actual);
    }

    @Override
    public final void onNext(T t) {
        if (isCancelled()) {
            return;
        }

        if (t == null) {
            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
            return;
        }
        
        //这里暂时将get()函数看成是相似于List这种的容器，存储的是当前须要处理的数据项
        if (get() != 0) { //从数据项容器中取值，若是当前有数据项须要处理
            actual.onNext(t); //发射数据项
            BackpressureHelper.produced(this, 1); //对当前存在须要处理的数据项进行-1操做
        } else {
            onOverflow(); //从数据项容器中取值，若是当前没有数据项须要处理，则回调溢出函数
        }
    }

    abstract void onOverflow();
}
复制代码

NoOverflowBaseAsyncEmitter在发射数据项的时候，会去BaseEmitter中的数据项容器去取出数据项，若是存在则处理，不存在则表示溢出，回调溢出函数，那么具体的数据项容器时候怎么存储须要处理的数据项的呢

六、BaseEmitter

abstract static class BaseEmitter<T>
    extends AtomicLong
    implements FlowableEmitter<T>, Subscription {
    
    private static final long serialVersionUID = 7326289992464377023L;

    final Subscriber<? super T> actual;

    final SequentialDisposable serial;

    BaseEmitter(Subscriber<? super T> actual) {
        this.actual = actual;
        this.serial = new SequentialDisposable();
    }

    @Override
    public void onComplete() {
        complete();
    }

    protected void complete() {
        if (isCancelled()) {
            return;
        }
        try {
            actual.onComplete();
        } finally {
            serial.dispose();
        }
    }

    @Override
    public final void onError(Throwable e) {
        if (!tryOnError(e)) {
            RxJavaPlugins.onError(e);
        }
    }

    @Override
    public final void request(long n) {
        if (SubscriptionHelper.validate(n)) {
            BackpressureHelper.add(this, n);
        }
    }
}
复制代码

BaseEmitter就是一个AtomicLong，若是没学过AtomicLong的话，能够简单理解为一个计数器，get()就是获取当前的Long值，只要不等于0就表示有值。主要仍是在request()，request()表示此时须要处理的数据项。结合上面NoOverflowBaseAsyncEmitter的中的BackpressureHelper.produced(this, 1)和当前BaseEmitter中的BackpressureHelper.add(this, n)，可得数据项的容器彻底都是由BackpressureHelper去控制，咱们只须要对BackpressureHelper的存储和获取作分析，就能够知道当前是否有数据项须要处理

七、BackpressureHelper

public final class BackpressureHelper {

    public static long add(AtomicLong requested, long n) {
        for (;;) {
            long r = requested.get(); //获取当前数据项
            if (r == Long.MAX_VALUE) {
                return Long.MAX_VALUE;
            }
            long u = addCap(r, n);//当前数据项 + 新增的数据项
            if (requested.compareAndSet(r, u)) { //设置最新的数据项
                return r;
            }
        }
    }
    
    public static long addCap(long a, long b) {
        long u = a + b;
        if (u < 0L) {
            return Long.MAX_VALUE;
        }
        return u;
    }
    
    public static long produced(AtomicLong requested, long n) {
        for (;;) {
            long current = requested.get(); //获取当前数据项
            if (current == Long.MAX_VALUE) {
                return Long.MAX_VALUE;
            }
            long update = current - n; //当前数据项 - 须要发射的数据项（从源码上，n为1）
            if (update < 0L) { //不能为负数
                RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
                update = 0L;
            }
            if (requested.compareAndSet(current, update)) { //设置最新的数据项
                return update;
            }
        }
    }
}
复制代码

BackpressureHelper就是利用AtomicLong的原子性就行简单的计数器操做而已，并无什么复杂的操做。至此，咱们就知道背压的原理原来就是利用AtomicLong计数器和生产消费的模式去决定是否发射当前的数据项而已

10.4 RxJava的经常使用操做符原理

知识点：

理解map操做符的原理

RxJava经常使用操做符的表明就是map，分析map源码后，其余的操做符的思想是同样的，只不过是实现逻辑不一致而已。下面咱们经过分析map的主要流程去分析map是如何转换字符串的，从上面咱们知道Observable的建立、订阅、发射的过程，此次对于重复的内容就再也不继续分析，主要是分析中间map是如何回调apply()去将数据项转换成字符串的

public void map() {
    //建立被观察者
    Observable
            .create(new ObservableOnSubscribe<String>() {
                @Override
                //默认在主线程里执行该方法
                public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                    e.onNext("俊俊俊很帅");
                    e.onNext("你值得拥有");
                    e.onNext("取消关注");
                    e.onNext("但仍是要保持微笑");
                    e.onComplete();
                }
            })
            .map(new Function<String, String>() {
                @Override
                public String apply(String s) throws Exception {
                    return "Hello";
                }
            })
            //建立观察者并订阅
            .subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
                    if (!d.isDisposed()) {
                        d.dispose();
                    }
                }

                @Override
                public void onNext(String s) {
                    System.out.println("onNext=" + s);
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("onNext=" + e.getMessage());
                }

                @Override
                public void onComplete() {

                }
            });
}
复制代码

一、Observable.map

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

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
    ObjectHelper.requireNonNull(mapper, "mapper is null");
    return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
复制代码

从create到map的过程当中，create的时候，当前的Observable已经被转换成ObservableCreate，再次map的时候，当前的Observable已经被转换成ObservableMap，并且在ObservableMap中传递的参数包含this，因此当前ObservableMap中是嵌套着ObservableCreate

二、Observable.subscribe

因为当前的Observable是ObservableMap，因此Observable.subscribe会回调ObservableMap中的subscribeActual

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends U> function;

    public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
        super(source);
        this.function = function;
    }

    @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MapObserver<T, U>(t, function));//source是传递进来的ObservableCreate
    }
}
复制代码

ObservableMap中的subscribeActual，会去调用ObservableCreate的subscribe方法，最后仍是会去回调 ObservableCreate的subscribeActual，不过这里在回调的过程当中增长了一个参数MapObserver，这个参数只有在ObservableCreate发射器发射的时候才会被调用

三、ObservableCreate.subscribeActual

@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);
    }
}
复制代码

在ObservableCreate.subscribeActual中，会接收一个Observer的参数，这个时候的Observer的参数是从ObservableMap中传递过来的MapObserver，当CreateEmitter发射onNext的时候，就会在当前的MapObserver对象onNext进行处理

四、MapObserver.onNext

static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
    final Function<? super T, ? extends U> mapper;

    MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
        super(actual);
        this.mapper = mapper;
    }

    @Override
    public void onNext(T t) {
        if (done) {
            return;
        }

        if (sourceMode != NONE) {
            actual.onNext(null);
            return;
        }

        U v;

        try {
            v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
        } catch (Throwable ex) {
            fail(ex);
            return;
        }
        actual.onNext(v);
    }
    
    ......
}
复制代码

onNext主要作了两个事情，一个是mapper.apply(t)，这个就是map操做符所实现的方法，这里就将原来的值转换成新的值，一个是actual.onNext(v)，将转换出来的新值v继续onNext出去，这里的actual就是在构造函数中传递进来的ObservableCreate，这里就已经将数据项通过map的操做符后继续执行后面正常的发射流程

小结

如图所示

10.5 RxJava的线程切换原理

知识点：

理解在工做线程上为何能执行耗时操做
理解在UI线程为何能执行更新UI的操做

沿用上面的例子，在线程切换的过程当中，无非就是至关于不一样的操做符继续操做数据项而已，根本的实现思路和map等操做符是同样的，也是经过嵌套Observable的过程来执行的，只不过是线程切换的操做符内部实现的逻辑有区别而已。经过咱们以往的思路去想，这两个知识点无非就是启动线程池去执行耗时任务，而UI线程则是交给Handler去处理，RxJava线程切换的原理就是这样的

Observable
        .create(new ObservableOnSubscribe<String>() {
            @Override
            //默认在主线程里执行该方法
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext("俊俊俊很帅");
                e.onNext("你值得拥有");
                e.onNext("取消关注");
                e.onNext("但仍是要保持微笑");
                e.onComplete();
            }
        })
        .map(new Function<String, String>() {
            @Override
            public String apply(String s) throws Exception {
                return "Hello";
            }
        })
        //将被观察者切换到子线程
        .subscribeOn(Schedulers.io())
        //将观察者切换到主线程  须要在Android环境下运行
        .observeOn(AndroidSchedulers.mainThread())
        //建立观察者并订阅
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                if (!d.isDisposed()) {
                    d.dispose();
                }
            }

            @Override
            public void onNext(String s) {
                System.out.println("onNext=" + s);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onNext=" + e.getMessage());
            }

            @Override
            public void onComplete() {

            }
        });
复制代码

基础概念：

Schedulers：调度器的管理者。管理着多种不一样种类的Scheduler
Scheduler：调度器。负责线程Worker的建立createWorker()，调度Worker的执行schedule()
Worker：抽象的工做线程。被线程调度器管理，负责线程的建立和执行

在源码中，咱们须要先熟悉这三者之间的关系究竟是如何运做的

一、observeOn()

@CheckReturnValue
@SchedulerSupport("custom")
public final Observable<T> observeOn(Scheduler scheduler) {
    return this.observeOn(scheduler, false, bufferSize());
}

@CheckReturnValue
@SchedulerSupport("custom")
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    ObjectHelper.verifyPositive(bufferSize, "bufferSize");
    return RxJavaPlugins.onAssembly(new ObservableObserveOn(this, scheduler, delayError, bufferSize));
}
复制代码

当前的Observable已经被转换成ObservableObserveOn

public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;
    final boolean delayError;
    final int bufferSize;

    public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
        super(source);
        this.scheduler = scheduler;
        this.delayError = delayError;
        this.bufferSize = bufferSize;
    }

    protected void subscribeActual(Observer<? super T> observer) {
        if (this.scheduler instanceof TrampolineScheduler) {
            this.source.subscribe(observer);
        } else {
            //一、建立工做线程
            Worker w = this.scheduler.createWorker();
            //二、订阅以后，在发射的过程当中
            this.source.subscribe(new ObservableObserveOn.ObserveOnObserver(observer, w, this.delayError, this.bufferSize));
        }

    }

    static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
        private static final long serialVersionUID = 6576896619930983584L;
        final Observer<? super T> actual;
        final Worker worker;
        final boolean delayError;
        final int bufferSize;
        SimpleQueue<T> queue;
        
        public void onNext(T t) {
            if (!this.done) {
                if (this.sourceMode != 2) {
                    this.queue.offer(t);
                }
                //三、在OnNext中执行
                this.schedule();
            }
        }

        void schedule() {
            if (this.getAndIncrement() == 0) {
                //四、执行工做线程
                this.worker.schedule(this);
            }
        }
    }
}
复制代码

其三者的关系简单的说就是在每次订阅的时候，都会去建立出对应的工做线程，这个工做线程取决于你传递的参数是哪一个Worker，在发射器发射的过程当中，这个工做线程总会去执行它的回调schedule，其实大部分的操做就是在schedule里面执行线程。搞懂了三者的关系以后，分析线程切换就简单多了，就至关于工厂同样，给个具体的任务给到具体的工人去执行，很像工厂的流水线，咱们已经肯定下来了流水线的流程了，这个时候咱们就须要去关心参数具体是什么东西了。在阅读subscribeOn、observeOn前，咱们先看看这两个方法中的参数都是什么

一、Schedulers.io()

public final class Schedulers {

    @NonNull
    static final Scheduler IO;
    
    static {
        
        SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask());

        COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask());
        
        //一、在初始化的时候就构建出了IOTask，initIoScheduler会去执行IOTask的call方法
        IO = RxJavaPlugins.initIoScheduler(new IOTask());

        TRAMPOLINE = TrampolineScheduler.instance();

        NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());
    }
    
    static final class IOTask implements Callable<Scheduler> {
        @Override
        public Scheduler call() throws Exception {
            //二、IOTask的call方法会去获取IoHolder的值
            return IoHolder.DEFAULT;
        }
    }
    
    static final class IoHolder {
        //三、建立IoScheduler
        static final Scheduler DEFAULT = new IoScheduler();
    }
    
    public static Scheduler io() {
        //Schedulers.io()：它会去获取前面3步建立出来的IoScheduler对象
        return RxJavaPlugins.onIoScheduler(IO); //返回IO自身
    }
}
复制代码

其正在实如今IoScheduler，其表示管理Io线程的管理者

public final class IoScheduler extends Scheduler {
    
    final AtomicReference<CachedWorkerPool> pool;
    
    static final CachedWorkerPool NONE;
    static {
        ......
        //一、建立CachedWorkerPool
        NONE = new CachedWorkerPool(0, null, WORKER_THREAD_FACTORY);
    }
    
    static final class CachedWorkerPool implements Runnable {
        
        ......
        private final ScheduledExecutorService evictorService;
        private final Future<?> evictorTask;
        
        CachedWorkerPool(long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) {
            ......
            Future<?> task = null;
            if (unit != null) {
                evictor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);
                task = evictor.scheduleWithFixedDelay(this, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS);
            }
            evictorService = evictor;
            evictorTask = task;
        }
    }
    
    @NonNull
    @Override
    public Worker createWorker() {
        //二、建立具体的线程
        return new EventLoopWorker(pool.get());
    }
    
    static final class EventLoopWorker extends Scheduler.Worker {
        private final CompositeDisposable tasks;
        private final CachedWorkerPool pool;
        private final ThreadWorker threadWorker;

        final AtomicBoolean once = new AtomicBoolean();

        EventLoopWorker(CachedWorkerPool pool) {
            this.pool = pool;
            this.tasks = new CompositeDisposable();
            this.threadWorker = pool.get();
        }

        @NonNull
        @Override
        public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
            if (tasks.isDisposed()) {
                // don't schedule, we are unsubscribed return EmptyDisposable.INSTANCE; } //三、最终会去调用ThreadWorker的scheduleActual return threadWorker.scheduleActual(action, delayTime, unit, tasks); } } //四、因为ThreadWorker没有scheduleActual，在父类中找NewThreadWorker static final class ThreadWorker extends NewThreadWorker { private long expirationTime; ThreadWorker(ThreadFactory threadFactory) { super(threadFactory); this.expirationTime = 0L; } public long getExpirationTime() { return expirationTime; } public void setExpirationTime(long expirationTime) { this.expirationTime = expirationTime; } } } 复制代码

NewThreadWorker，最终仍是调用executor.submit()或executor.schedule()

@NonNull
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable 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) {
        if (parent != null) {
            parent.remove(sr);
        }
        RxJavaPlugins.onError(ex);
    }

    return sr;
}
复制代码

二、AndroidSchedulers.mainThread()

public final class AndroidSchedulers {

    private static final class MainHolder {

        static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
    }

    private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
            new Callable<Scheduler>() {
                @Override public Scheduler call() throws Exception {
                    return MainHolder.DEFAULT;
                }
            });

    /** A {@link Scheduler} which executes actions on the Android main thread. */
    public static Scheduler mainThread() {
        return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
    }
}
复制代码

返回一个HandlerScheduler，建立单例模式的主线程Handler

final class HandlerScheduler extends Scheduler {
    private final Handler handler;

    HandlerScheduler(Handler handler) {
        this.handler = handler;
    }

    public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
        if (run == null) {
            throw new NullPointerException("run == null");
        } else if (unit == null) {
            throw new NullPointerException("unit == null");
        } else {
            run = RxJavaPlugins.onSchedule(run);
            HandlerScheduler.ScheduledRunnable scheduled = new HandlerScheduler.ScheduledRunnable(this.handler, run);
            this.handler.postDelayed(scheduled, unit.toMillis(delay));
            return scheduled;
        }
    }

    public Worker createWorker() {
        //建立具体工做线程
        return new HandlerScheduler.HandlerWorker(this.handler);
    }

    ......
}
复制代码

就好像咱们上面分析的三者关系同样，Schedule最终仍是会管理着具体的工做线程

private static final class HandlerWorker extends Worker {
    private final Handler handler;

    private volatile boolean disposed;

    HandlerWorker(Handler handler) {
        this.handler = handler;
    }

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

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

        run = RxJavaPlugins.onSchedule(run);
        //包装新的Runnable交给Handler
        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; } @Override public void dispose() { disposed = true; handler.removeCallbacksAndMessages(this /* token */); } @Override public boolean isDisposed() { return disposed; } } 复制代码

三、subscribeOn()

理解完参数后，回到咱们的分析重点

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

subscribeOn就如同普通操做符同样，包装一层ObservableSubscribeOn，在subscribe的时候真正走的仍是subscribeActual

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    @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 SubscribeTask(parent)));
    }
    
    final class SubscribeTask implements Runnable {
        private final SubscribeOnObserver<T> parent;
    
        SubscribeTask(SubscribeOnObserver<T> parent) {
            this.parent = parent;
        }
    
        @Override
        public void run() {
            //三、回调ObservableOnSubscribe的subscribe
            source.subscribe(parent);
        }
    }
}
复制代码

scheduler.scheduleDirect中会去执行Scheduler里的方法，这里的scheduler就是IoScheduler

@NonNull
public Disposable scheduleDirect(@NonNull Runnable run) {
    return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}

@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;
}
复制代码

回调IoScheduler的createWorker()并执行w.schedule()

小结

如图所示

10.6 RxJava的自定义Operator原理

知识点：

自定义Operator是如何实现的

在讲解以前，让咱们先回味下自定义Operator

public class CustomOperator implements ObservableOperator<String, List<String>> {

    @Override
    public Observer<? super List<String>> apply(final Observer<? super String> observer) throws Exception {
        return new Observer<List<String>>() {
            @Override
            public void onSubscribe(Disposable d) {
                observer.onSubscribe(d);
            }

            @Override
            public void onNext(List<String> strings) {
                observer.onNext(strings.toString());
            }

            @Override
            public void onError(Throwable e) {
                observer.onError(e);
            }

            @Override
            public void onComplete() {
                observer.onComplete();
            }
        };
    }
}
复制代码

Observable.create(new ObservableOnSubscribe<List<String>>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<List<String>> e) throws Exception {
      
    }
}).lift(new CustomOperator())
复制代码

自定义Operator如同普通的操做符原理差很少，用的是lift的操做符，只不过在lift里面将逻辑的执行回调到自定义的Operator的apply()

一、Observable.lift

public final <R> Observable<R> lift(ObservableOperator<? extends R, ? super T> lifter) {
    ObjectHelper.requireNonNull(lifter, "onLift is null");
    return RxJavaPlugins.onAssembly(new ObservableLift<R, T>(this, lifter));
}
复制代码

二、ObservableLift.subscribeActual

public final class ObservableLift<R, T> extends AbstractObservableWithUpstream<T, R> {
    /** The actual operator. */
    final ObservableOperator<? extends R, ? super T> operator;

    public ObservableLift(ObservableSource<T> source, ObservableOperator<? extends R, ? super T> operator) {
        super(source);
        this.operator = operator;
    }

    @Override
    public void subscribeActual(Observer<? super R> s) {
        Observer<? super T> observer;
        try {
            observer = ObjectHelper.requireNonNull(operator.apply(s), "Operator " + operator + " returned a null Observer");
        } 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 Disposable already
            RxJavaPlugins.onError(e);

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

        source.subscribe(observer);
    }
}
复制代码

能够看到代码很是快的就将传递进来的参数operator执行apply()

10.7 RxJava的自定义Transformer原理

知识点：

自定义Transformer是如何实现的

在讲解以前，让咱们先回味下自定义Transformer

public class NetWorkTransformer implements ObservableTransformer {

    @Override
    public ObservableSource apply(Observable upstream) {
        return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }
}
复制代码

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        
    }
}).compose(new CustomTransformer())
复制代码

自定义Transformer如同普通的操做符原理差很少，用的是compose的操做符，只不过在compose里面将逻辑的执行回调到自定义的Transformer的apply()

一、Observable.compose

public final <R> Observable<R> compose(ObservableTransformer<? super T, ? extends R> composer) {
    return wrap(((ObservableTransformer<T, R>) ObjectHelper.requireNonNull(composer, "composer is null")).apply(this));
}
复制代码

能够看到代码很是快的就将传递进来的参数composer执行apply()，这里的wrap()只是将代码裹了一层，若是你想简单的理解的话，能够理解为做者的强迫症犯了，只是为了让全部代码看起来都比较规范，否则这里实在和其余操做符的实现不同，咱们能够追进去wrap()

public static <T> Observable<T> wrap(ObservableSource<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    if (source instanceof Observable) {
        return RxJavaPlugins.onAssembly((Observable<T>)source);
    }
    return RxJavaPlugins.onAssembly(new ObservableFromUnsafeSource<T>(source));
}
复制代码

wrap()实际上是对composer操做符作了Hook，由于全部操做符都会被RxJava去Hook住，这里会在下面讲到自定义Plugin原理的时候就明白了

10.8 RxJava的自定义Plugin原理

知识点：

自定义Plugin是如何实现AOP的

在讲解以前，让咱们先回味下自定义Plugin

public class CustomObservableAssembly implements Function<Observable, Observable> {
    @Override
    public Observable apply(Observable observable) throws Exception {
        System.out.println("CustomObservableAssembly observable.toString:" + observable.toString());
        return observable;
    }
}
复制代码

RxJavaPlugins.setOnObservableAssembly(new CustomObservableAssembly());
复制代码

在自定义Plugin中，相似于Android的术语Hook，但在这里并非真正的Hook，而是做者在写RxJava的时候去限定一套规范，让后面的全部操做符或其余操做等，均可以实现Hook的原理

一、RxJavaPlugins.setOnObservableAssembly

public static void setOnObservableAssembly(@Nullable Function<? super Observable, ? extends Observable> onObservableAssembly) {
    if (lockdown) {
        throw new IllegalStateException("Plugins can't be changed anymore");
    }
    RxJavaPlugins.onObservableAssembly = onObservableAssembly;
}
复制代码

RxJavaPlugins.setOnObservableAssembly只是对成员变量设置了自定义的值，这个时候onObservableAssembly就有了值，默认是为null的。设置完值就表示已经Hook成功了，当操做符执行的时候，是如何回调咱们Hook的函数的

二、Observable.create

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
复制代码

create至关于一个操做符，在每一个操做符的里面都会去执行一段RxJavaPlugins.onAssembly，这里就是RxJava规定的规范，一开始咱们只是说返回自身，可是有了Hook以后，就会回调Hook函数，返回已经通过二次加工的自身

三、RxJavaPlugins.onAssembly

public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
    Function<? super Observable, ? extends Observable> f = onObservableAssembly;
    if (f != null) {
        return apply(f, source);
    }
    return source;
}
复制代码

因为咱们已经设置了新值，这里的onObservableAssembly就不为null，不为null则执行apply()，apply()就是咱们Hook传进去参数的回调方法

10.9 美团WhiteBoard

美团的WhiteBoard实际上是取自美团的开源框架Shield——开源的移动端页面模块化开发框架中的代码，其主要做用是应用RxJava的Subject搭起组件间通信的桥梁。实质上在WhiteBoard中，是将全部的组件的数据和Subject通信的桥梁保存起来，经过key做为组件的惟一标志。不过比较惋惜的是WhiteBoard使用的是RxJava1，不过关系不大，只要读懂里面的源码便可

一、WhiteBoard的初始化

初始化放在Activity/Fragment界面中，至关于通信的桥梁，每一个界面中仅有一个WhiteBoard的实例，并由全部组件共用

public abstract class ShieldFragment extends Fragment implements AgentCellBridgeInterface, DriverInterface {
    
    static final String TAG = ShieldFragment.class.getSimpleName();
    ......
    protected WhiteBoard whiteBoard;

    public ShieldFragment() {
        this.whiteBoard = new WhiteBoard();//初始化
    }

    @Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        ......
        whiteBoard.onCreate(savedInstanceState);//对应生命周期
    }

    @Override
    public void onDestroy() {
        super.onDestroy();
        ......
        whiteBoard.onDestory();//对应生命周期
    }

    @Override
    public void onSaveInstanceState(Bundle outState) {
        super.onSaveInstanceState(outState);
        ......
        whiteBoard.onSaveInstanceState(outState);//对应生命周期
    }

    @Override
    public WhiteBoard getWhiteBoard() {
        return whiteBoard;//获取实例
    }
}
复制代码

二、WhiteBoard监听通知

组件只监听某个key的事件，有通知的时候就能收到

public class MixCellAgent extends LightAgent {
    private MixCell mixCell;
    private Subscription loadingSubscription;
    private Subscription emptySubscription;

    public MixCellAgent(Fragment fragment, DriverInterface bridge, PageContainerInterface pageContainer) {
        super(fragment, bridge, pageContainer);
    }

    @Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        mixCell = new MixCell(getContext(), this);
        loadingSubscription = getWhiteBoard().getObservable(MixLoadingAgent.KEY_LOADING).filter(new Func1() {
            @Override
            public Object call(Object o) {
                return o instanceof Boolean && ((Boolean) o);
            }
        }).subscribe(new Action1() {
            @Override
            public void call(Object o) {
                loading();
            }
        });

        emptySubscription = getWhiteBoard().getObservable(MixLoadingAgent.KEY_EMPTY).filter(new Func1<Object, Boolean>() {
            @Override
            public Boolean call(Object o) {
                return o instanceof Boolean && ((Boolean) o);
            }
        }).subscribe(new Action1() {
            @Override
            public void call(Object o) {
                mixCell.onEmpty();
            }
        });
        
        ......
    }

    @Override
    public void onDestroy() {
        if (loadingSubscription != null) {
            loadingSubscription.unsubscribe();
            loadingSubscription = null;
        }

        if (emptySubscription != null) {
            emptySubscription.unsubscribe();
        }
        
        ......
    }
}
复制代码

三、WhiteBoard的发送通知

WhiteBoard发送通知就是调用WhiteBoard提供的全部put方法，具体是如何收到消息的，还须要经过WhiteBoard的源码看下

public class MixLoadingAgent extends LightAgent implements MixLoadingCell.MixLoadingListener {
    public static final String KEY_LOADING = "loading";
    public static final String KEY_EMPTY = "empty";
    public static final String KEY_FAILED = "failed";
    public static final String KEY_MORE = "more";
    public static final String KEY_DONE = "done";

    private MixLoadingCell mixLoadingCell;

    public MixLoadingAgent(Fragment fragment, DriverInterface bridge, PageContainerInterface pageContainer) {
        super(fragment, bridge, pageContainer);
        mixLoadingCell = new MixLoadingCell(getContext());
        mixLoadingCell.setOnMixLoadingListener(this);
    }

    @Override
    public SectionCellInterface getSectionCellInterface() {
        return mixLoadingCell;
    }

    @Override
    public void onLoading() {
        getWhiteBoard().putBoolean(KEY_LOADING, true);
    }

    @Override
    public void onEmpty() {
        getWhiteBoard().putBoolean(KEY_EMPTY, true);
    }

    @Override
    public void onFailed() {
        getWhiteBoard().putBoolean(KEY_FAILED, true);
    }

    @Override
    public void onMore() {
        getWhiteBoard().putBoolean(KEY_MORE, true);
    }

    @Override
    public void onDone() {
        getWhiteBoard().putBoolean(KEY_DONE, true);
    }
}
复制代码

四、WhiteBoard的原理

最后只须要获取实例后发送通知便可，getWhiteBoard().putBoolean(key)。WhiteBoard原理是只要仍是Subject的桥梁的做用

public class WhiteBoard {

    public static final String WHITE_BOARD_DATA_KEY = "White_Board_Data";
    protected Bundle mData;//保存全部组件的数据
    protected HashMap<String, Subject> subjectMap;//保存全部组件的通信桥梁

    public WhiteBoard() {
        this(null);
    }

    public WhiteBoard(Bundle data) {
        mData = data;
        if (mData == null) {
            mData = new Bundle();//初始化
        }

        subjectMap = new HashMap<>();//初始化
    }

    public void onCreate(Bundle savedInstanceState) {
        if (savedInstanceState != null) {
            mData = savedInstanceState.getBundle(WHITE_BOARD_DATA_KEY);
        }

        if (mData == null) {
            mData = new Bundle();
        }
    }

    public void onSaveInstanceState(Bundle outState) {
        if (outState != null) {

            // here we must save a new copy of the mData into the outState
            outState.putBundle(WHITE_BOARD_DATA_KEY, new Bundle(mData));
        }
    }

    public void onDestory() {
        subjectMap.clear();
        mData.clear();
    }

    //经过key获取某组件的桥梁
    public Observable getObservable(final String key) {

        Subject res = null;
        if (subjectMap.containsKey(key)) {
            res = subjectMap.get(key);
        } else {
            res = PublishSubject.create();
            subjectMap.put(key, res);
        }
        if (getData(key) != null) {
            return res.startWith(getData(key));//带上已经存储过的数据
        } else {
            return res;
        }
    }

    //经过key通知某组件
    protected void notifyDataChanged(String key) {
        if (subjectMap.containsKey(key)) {
            subjectMap.get(key).onNext(mData.get(key));
        }
    }

    //移除组件中的数据
    public void removeData(String key) {
        mData.remove(key);
        notifyDataChanged(key);
    }
    
    //每次put值的时候，就会去通知对应的组件
    public void putBoolean(@Nullable String key, boolean value) {
        mData.putBoolean(key, value);
        notifyDataChanged(key);
    }
    public void putInt(@Nullable String key, int value) {
        mData.putInt(key, value);
        notifyDataChanged(key);
    }
    public void putString(@Nullable String key, @Nullable String value) {
        mData.putString(key, value);
        notifyDataChanged(key);
    }
    ......

    public double getDouble(String key) {
        return mData.getDouble(key);
    }
    public String getString(String key, String defaultValue) {
        return mData.getString(key, defaultValue);
    }
    ......
}
复制代码

1. 第一章课程介绍
2. 0课程介绍（Week1，3月3日）
3. Maven工程目录介绍
4. Keras 入门课0 -- 目录
5. 课程介绍
6. Angular项目目录介绍
7. 小程序开发第二课——工程文件目录介绍
8. laravel目录介绍
9. CentOs目录介绍
10. CentOS 目录介绍
更多相关文章...
• Spring目录结构和基础JAR包介绍 - Spring教程
• 网站主机介绍 - 网站主机教程
• Java Agent入门实战（一）-Instrumentation介绍与使用
• Java 8 Stream 教程