RXJava 原理分析

一、RxJava 简介

RxJava是一种基于观察者模式的响应式编程框架，其定义的角色有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]");
    }
});

运行结果：

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> {    
    // 创建事件源
    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

    // 订阅相关
    public final void subscribe(Observer<? super T> observer) {
        // ...
        subscribeActual(observer);
    }
    
    protected abstract void subscribeActual(Observer<? super T> observer);
}

create方法创建事件源,其参数ObservableOnSubscribe 接口定义如下：
ObservableOnSubscribe.java

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

ObservableEmitter可以理解为事件发射器，继承Emitter里卖弄的基本发射方法，同时关连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方法最终会返回ObservableCreate<T>对象，ObservableCreate类包含了ObservableOnSubscribe对象引用，且实现了subscribeActual订阅方法

ObservableCreate.java

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) {
        // 1 创建 CreateEmitter，也是一个适配器，可以将 Observer -> Disposable，CreateEmitter 中主要持有 observer 对象的引用，并且维护了 dispose 变量。
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        //2 onSubscribe（）参数是 Disposable。还有一点要注意的是 onSubscribe() 是在我们执行 subscribe() 这句代码的那个线程回调的，并不受线程调度影响。
        // 给 observer 的一个回调，告诉它是否 dispose
        observer.onSubscribe(parent);

        try {
            //3 将 ObservableOnSubscribe（源头）与 CreateEmitter（Observer，终点）联系起来，即完成订阅，此时 ObservableOnSubscribe 会向 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，会调用 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() 互斥只能执行一次
            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函数的源码

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变换函数类保存起来
        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以后才会是true，默认这里是false，所以跳过
            if (done) {
                return;
            }
            // 默认sourceMode是0，所以跳过
            if (sourceMode != NONE) {
                downstream.onNext(null);
                return;
            }

            U v;

            try {
                // 这一步执行变换,将上游传过来的 T，利用 Function 转换成下游需要的 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 线程调度

subscribeOn 函数源码：

public final Observable<T> subscribeOn(Scheduler scheduler) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    return RxJavaPlugins.onAssembly(
        // 返回Observable，传入上游this和调度器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实现了Disposable，将其传递给下游的onSubscribe
        downStream.onSubscribe(parent);
        // 把source的订阅放在Runnable中，由scheduler调度
        parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
            @Override
            public void run() {
                source.subscribe(parent);
            }
        }));
    }
}

observeOn函数源码:

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; // 线程调度器
        this.delayError = delayError; // 出现错误是否立刻中断线程
        this.bufferSize = bufferSize;   //缓冲区大小，默认128
    }
    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            // 默认线程则不做线程调度，直接在当前线程中调用
            source.subscribe(observer);
        } else {
            Scheduler.Worker w = scheduler.createWorker();
            source.subscribe(
                // 用Worker、和相关参数装饰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;
                //省略部分代码，创建缓冲队列
                queue = new SpscLinkedArrayQueue<T>(bufferSize);
                actual.onSubscribe(this);
            }
    }
    
    @Override
    public void onNext(T t) {
      if (done) {
        return;
      }
      if (sourceMode != QueueDisposable.ASYNC) {
        queue.offer(t); //上游的数据全部先入队列
      }
      //执行调度
      schedule();
    }
    
    void schedule() {
      if (getAndIncrement() == 0) {
        // 队列如果已经空了，则再次调度
        worker.schedule(this);
      }
    }
    
    @Override
    public void run() {
        // Fused 熔断机制，默认false
        if (outputFused) {
            drainFused();
        } else {
            drainNormal();
        }
    }
    
    //该函数在Runnable所在的线程执行，从缓冲队列里拿出事件，向下游发射
    void drainNormal() {
        int missed = 1;
        final SimpleQueue<T> q = queue;
        final Observer<? super T> a = actual;
        for (;;) {
            // 如果设置了errorDelay，则不管队列是否为空，发生了错误都会中断发射，并调用observer的onError
            if (checkTerminated(done, q.isEmpty(), a)) {
                return;
            }
            for (;;) {
                boolean d = done;
                T v;
                try {
                    v = q.poll();   //队列中取数据
                } 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;
                }
                // 向下游发射数据
                a.onNext(v);
            }
            missed = addAndGet(-missed);
            if (missed == 0) {
                break;
            }
        }
    }
  }
}

四、更多阅读

1. RxJava 原理篇