一、RxJava 簡(jiǎn)介

RxJava是一種基于觀察者模式的響應(yīng)式編程框架，其定義的角色有Observable事件流、ObservableOnSubscribe事件源頭、Observer是事件訂閱者。

二、基本使用

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("Android");
        emitter.onNext("Kotlin");
        emitter.onNext("Java");
        emitter.onComplete();
    }
}).map(new Function<String, String>() {
    @Override
    public String apply(String s) throws Exception {
        return "I Love "+ s;
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) {
    }
    
    @Override
    public void onNext(String s) {
        Log.d("test", "onNext: "+s);
    }

    @Override
    public void onError(Throwable e) {
    }

    @Override
    public void onComplete() {
        Log.d("test", "[onComplete]");
    }
});

運(yùn)行結(jié)果：

test: onNext: I Love Android
test: onNext: I Love Kotlin
test: onNext: I Love Java
test: [onComplete]

三、原理剖析

3.1 Observable、Observer訂閱原理

Observable.java

public abstract class Observable<T> implements ObservableSource<T> {    
    // 創(chuàng)建事件源
    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

    // 訂閱相關(guān)
    public final void subscribe(Observer<? super T> observer) {
        // ...
        subscribeActual(observer);
    }
    
    protected abstract void subscribeActual(Observer<? super T> observer);
}

create方法創(chuàng)建事件源,其參數(shù)ObservableOnSubscribe 接口定義如下：
ObservableOnSubscribe.java

public interface ObservableOnSubscribe<T> {
    void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}

ObservableEmitter可以理解為事件發(fā)射器，繼承Emitter里賣弄的基本發(fā)射方法，同時(shí)關(guān)連Disposable
Emitter.java

public interface Emitter<T> {
    void onNext(@NonNull T value);
    void onError(@NonNull Throwable error);
    void onComplete();
}

ObservableEmitter.java

public interface ObservableEmitter<T> extends Emitter<T> {
    void setDisposable(@Nullable Disposable d);
    void setCancellable(@Nullable Cancellable c);
    boolean isDisposed();
    ObservableEmitter<T> serialize();
    boolean tryOnError(@NonNull Throwable t);
}

Observable.create方法最終會(huì)返回ObservableCreate<T>對(duì)象，ObservableCreate類包含了ObservableOnSubscribe對(duì)象引用，且實(shí)現(xiàn)了subscribeActual訂閱方法

ObservableCreate.java

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

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }
    
    // 訂閱的時(shí)候會(huì)執(zhí)行該方法
    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        // 1 創(chuàng)建 CreateEmitter，也是一個(gè)適配器，可以將 Observer -> Disposable，CreateEmitter 中主要持有 observer 對(duì)象的引用，并且維護(hù)了 dispose 變量。
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        //2 onSubscribe（）參數(shù)是 Disposable。還有一點(diǎn)要注意的是 onSubscribe() 是在我們執(zhí)行 subscribe() 這句代碼的那個(gè)線程回調(diào)的，并不受線程調(diào)度影響。
        // 給 observer 的一個(gè)回調(diào)，告訴它是否 dispose
        observer.onSubscribe(parent);

        try {
            //3 將 ObservableOnSubscribe（源頭）與 CreateEmitter（Observer，終點(diǎn)）聯(lián)系起來，即完成訂閱，此時(shí) ObservableOnSubscribe 會(huì)向 observer 傳送事件
            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;
            }
            // 如果沒有被 dispose，會(huì)調(diào)用 Observer 的 onNext()方法
            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() {
            // onComplete() 和 onError() 互斥只能執(zhí)行一次
            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());
        }

        @Override
        public String toString() {
            return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
        }
    }
}

其中Observer訂閱者定義如下：

public interface Observer<T> {
    void onSubscribe(@NonNull Disposable d);
    void onNext(@NonNull T t);
    void onError(@NonNull Throwable e);
    void onComplete();
}

3.2 map操作符原理

map函數(shù)的源碼

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

ObservableMap.java

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()將上游的Observable保存起來 ，用于subscribeActual()中用。
        super(source);
        // 將function變換函數(shù)類保存起來
        this.function = function;
    }

    @Override
    public void subscribeActual(Observer<? super U> t) {
        // 用 MapObserver 訂閱上游 Observable。
        source.subscribe(new MapObserver<T, U>(t, function));
    }

    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) {
            // done在onError 和 onComplete以后才會(huì)是true，默認(rèn)這里是false，所以跳過
            if (done) {
                return;
            }
            // 默認(rèn)sourceMode是0，所以跳過
            if (sourceMode != NONE) {
                downstream.onNext(null);
                return;
            }

            U v;

            try {
                // 這一步執(zhí)行變換,將上游傳過來的 T，利用 Function 轉(zhuǎn)換成下游需要的 V
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            // 變換后傳遞給下游Observer
            downstream.onNext(v);
        }

        @Override
        public int requestFusion(int mode) {
            return transitiveBoundaryFusion(mode);
        }

        @Nullable
        @Override
        public U poll() throws Exception {
            T t = qd.poll();
            return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
        }
    }    
}

3.3 subscribeOn 線程調(diào)度

subscribeOn 函數(shù)源碼：

public final Observable<T> subscribeOn(Scheduler scheduler) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    return RxJavaPlugins.onAssembly(
        // 返回Observable，傳入上游this和調(diào)度器scheduler
        new ObservableSubscribeOn<T>(this, scheduler));
}
    
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> downStream) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(downStream);
        // SubscribeOnObserver實(shí)現(xiàn)了Disposable，將其傳遞給下游的onSubscribe
        downStream.onSubscribe(parent);
        // 把source的訂閱放在Runnable中，由scheduler調(diào)度
        parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
            @Override
            public void run() {
                source.subscribe(parent);
            }
        }));
    }
}

observeOn函數(shù)源碼:

public final Observable<T> observeOn(Scheduler scheduler) {
    return observeOn(scheduler, false, bufferSize());
}
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<T>(this, scheduler, delayError, bufferSize));
}
// 以上代碼都很熟悉了，暫不贅述
 
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; // 線程調(diào)度器
        this.delayError = delayError; // 出現(xiàn)錯(cuò)誤是否立刻中斷線程
        this.bufferSize = bufferSize;   //緩沖區(qū)大小，默認(rèn)128
    }
    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            // 默認(rèn)線程則不做線程調(diào)度，直接在當(dāng)前線程中調(diào)用
            source.subscribe(observer);
        } else {
            Scheduler.Worker w = scheduler.createWorker();
            source.subscribe(
                // 用Worker、和相關(guān)參數(shù)裝飾observer，得到新的Observer注入上游
                new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }
  
  static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
    //省略部分代碼
    @Override
        public void onSubscribe(Disposable s) {
            if (DisposableHelper.validate(this.s, s)) {
                this.s = s;
                //省略部分代碼，創(chuàng)建緩沖隊(duì)列
                queue = new SpscLinkedArrayQueue<T>(bufferSize);
                actual.onSubscribe(this);
            }
    }
    
    @Override
    public void onNext(T t) {
      if (done) {
        return;
      }
      if (sourceMode != QueueDisposable.ASYNC) {
        queue.offer(t); //上游的數(shù)據(jù)全部先入隊(duì)列
      }
      //執(zhí)行調(diào)度
      schedule();
    }
    
    void schedule() {
      if (getAndIncrement() == 0) {
        // 隊(duì)列如果已經(jīng)空了，則再次調(diào)度
        worker.schedule(this);
      }
    }
    
    @Override
    public void run() {
        // Fused 熔斷機(jī)制，默認(rèn)false
        if (outputFused) {
            drainFused();
        } else {
            drainNormal();
        }
    }
    
    //該函數(shù)在Runnable所在的線程執(zhí)行，從緩沖隊(duì)列里拿出事件，向下游發(fā)射
    void drainNormal() {
        int missed = 1;
        final SimpleQueue<T> q = queue;
        final Observer<? super T> a = actual;
        for (;;) {
            // 如果設(shè)置了errorDelay，則不管隊(duì)列是否為空，發(fā)生了錯(cuò)誤都會(huì)中斷發(fā)射，并調(diào)用observer的onError
            if (checkTerminated(done, q.isEmpty(), a)) {
                return;
            }
            for (;;) {
                boolean d = done;
                T v;
                try {
                    v = q.poll();   //隊(duì)列中取數(shù)據(jù)
                } catch (Throwable ex) {
                    Exceptions.throwIfFatal(ex);
                    s.dispose();
                    q.clear();
                    a.onError(ex);
                    worker.dispose();
                    return;
                }
                boolean empty = v == null;
                if (checkTerminated(d, empty, a)) {
                    return;
                }
                if (empty) {
                    break;
                }
                // 向下游發(fā)射數(shù)據(jù)
                a.onNext(v);
            }
            missed = addAndGet(-missed);
            if (missed == 0) {
                break;
            }
        }
    }
  }
}

四、更多閱讀

1. RxJava 原理篇