一、框架思想

• 观察者模式

  • 观察者自下而上注入被观察者

  • 被观察者自上而下发射事件

    
    观察者模式

• 装饰器模式

  • 自上而下，被观察者被一层层装饰

  • 自下而上，观察者被一层层装饰

    
    装饰者模式

• 策略模式

  • 函数式编程

二、最简单的订阅模型

以下是由just方法和subscribe方法组成的一个最简单的订阅模型

Observable
    .just("Hello RxJava")
    .subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onNext(String s) {
            Log.d(TAG, s);
        }
        @Override
        public void onError(Throwable e) {
        }
        @Override
        public void onComplete() {
        Log.d(TAG, "Complete");
        }
    });

对于just方法

public static <T> Observable<T> just(T item) {
    // 参数合法判断，不允许传入空item
    ObjectHelper.requireNonNull(item, "The item is null");
    // hook装饰
    return RxJavaPlugins.onAssembly(
        // 返回真正的ObservableJust对象
        new ObservableJust<T>(item));
}
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
    Function<? super Observable, ? extends Observable> f = onObservableAssembly;
    if (f != null) {
        // 如果hook方法存在，则调用hook方法
        return apply(f, source);
    }
    return source;
}

public final class ObservableJust<T> extends Observable<T> implements ScalarCallable<T> {
    private final T value;
    public ObservableJust(final T value) {
        // 保存原始的value
        this.value = value;
    }
    
    // 实际的订阅方法的实现
    @Override
    protected void subscribeActual(Observer<? super T> downStream) {
        // 传入下游和value
        ScalarDisposable<T> sd = new ScalarDisposable<T>(downStream, value);
        // 订阅
        downStream.onSubscribe(sd);
        // Runnable.run()；
        sd.run();
    }
}

public static final class ScalarDisposable<T> extends AtomicInteger implements QueueDisposable<T>, Runnable {
    // 省略很多跟队列使用，线程安全相关的方法
    @Override
    public void run() {
        if (get() == START && compareAndSet(START, ON_NEXT)) {
            downStream.onNext(value);
            if (get() == ON_NEXT) {
                lazySet(ON_COMPLETE);
                downStream.onComplete();
            }
        }
    }
}

对于subscribe方法

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
        Action onComplete, Consumer<? super Disposable> onSubscribe) {
    // 参数合法检查
    ObjectHelper.requireNonNull(onNext, "onNext is null");
    ObjectHelper.requireNonNull(onError, "onError is null");
    ObjectHelper.requireNonNull(onComplete, "onComplete is null");
    ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
    // 把4个Lambda表达式合并创建LambdaObserver
    LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
    // 执行订阅
    subscribe(ls);
    // 返回
    return ls;
}
    
@Override
public final void subscribe(Observer<? super T> downStream) {
    // 检查参数合法性
    ObjectHelper.requireNonNull(downStream, "observer is null");
    try {
        // hook
        downStream = RxJavaPlugins.onSubscribe(this, downStream);
        // hook后再次检查Observer不为null
        ObjectHelper.requireNonNull(downStream, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
        // 执行实际的subscribe，传入下游Observer
        subscribeActual(downStream);
    } 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;
    }
}

三、操作符的实现

最简单的操作符 map

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
    // 检查参数合法性
    ObjectHelper.requireNonNull(mapper, "mapper is null");
    // hook
    return RxJavaPlugins.onAssembly(
        // 返回真正的 ObservableMap，传入了上游引用和mapper方法
        new ObservableMap<T, R>(this, mapper));
}

ObservableMap等操作符一般继承AbstractObservableWithUpstream，这种Observable持有了上游引用

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> downStream) {
        // 合并下游和mapper，创建新的Observer：MapObserver
        source.subscribe(new MapObserver<T, U>(downStream, function));
    }
    
    static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
        final Function<? super T, ? extends U> mapper;
        MapObserver(Observer<? super U> downStream, Function<? super T, ? extends U> mapper) {
            super(downStream);
            this.mapper = mapper;
        }
        
        // 重新实现onNext
        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            if (sourceMode != NONE) {
                downStream.onNext(null);
                return;
            }
            U v;
            try {
                // 检查mapper的输出结果，合法则赋值给
                v = ObjectHelper.requireNonNull(
                    mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            // 下游onNext传入经过mapper的结果
            downStream.onNext(v);
        }
    }

对于操作符而言

• 自上而下
  • 装饰器模式。操作符创建了一个新的 Observable
  • 该 Observable 重写了subscribeActual方法
  • subscribeActual 的实现是让所持有的上游source调用subscribe方法去订阅 被重新装饰的下游
• 自下而上
  • 装饰器模式。操作符使下游Observer被装饰后形成新的Observer
  • 重写了Observer的几个方法，向下传递经过操作符处理后的数值结果

自定义操作符

// compose()操作符，自上而下，封装Observable
public interface ObservableTransformer<Upstream, Downstream> {
    ObservableSource<Downstream> apply(Observable<Upstream> upstream);
}
// lift()操作符，自下而上，封装Observer
public interface ObservableOperator<Downstream, Upstream> {
    Observer<? super Upstream> apply(@NonNull Observer<? super Downstream> observer) throws Exception;
}

• lift
  • 创建并返回新的Observable，即ObservableLift
  • 订阅发生在operator.apply(s)，即得到新的Observable之后
  • 意图是封装一个操作符，类似于create
• compose
  • 意图是封装一系列操作符，方便复用

四、线程调度

举例 .subscribeOn(AndroidSchedulers.mainThread())

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

讲解

• subscribeOn 线程调度即发生在subscribe时
• subscribeOn 只能生效一次
  • 因为完整的订阅过程是自下而上订阅，数据源发射事件在自上而下传递，所以真正发射事件所在的线程，是有最接近上游的一次subscribeOn来决定的，其他的都会被覆盖
  • 从代码上看，就是经历了n个线程的传递后，把source.subscribe(parent)放在了最上游的那个线程中去发射，如果没有observeOn影响，整个事件流都会在那个线程完成
  • 为什么flatMap能改变subscribeOn的这种特性？因为flatMap等操作符创建了新的Observable，而不是单纯传递上下游

对于调度器 AndroidSchedulers.mainThread()。其实际上是 new HandlerScheduler(new Handler(Looper.getMainLooper()))

从上面看出，实现线程调度的是scheduler.scheduleDirect方法

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

    run = RxJavaPlugins.onSchedule(run);
    // ScheduledRunnable 实现了Disposable接口
    ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
    // 大家熟悉的handler.postDelayed
    handler.postDelayed(scheduled, Math.max(0L, unit.toMillis(delay)));
    return scheduled;
}

// 一般的Schedule都是用createWorker().schedule(scheduled, delay, unit)的方式去实现线程调度
// Worker实现相应的disposable接口，便于取消订阅时停止执行尚未执行的Runnable，并装饰相应的hook。
// 这个类的实现比较有点划水，可能因为是主线程，所以不担心泄露
@Override
public Worker createWorker() {
    return new HandlerWorker(handler);
}

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

        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, Math.max(0L, 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;
    }
}

举例 .observeOn(AndroidSchedulers.mainThread())

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

讲解

• 订阅发生在调度前，说明线程调度不影响订阅过程
• ObserveOnObserver持有下游observer和调度器，并实现Runnable接口
• 订阅时（onSubscribe）会创建一个缓冲队列，当上游数据到来先放在队列，接着在调度线程中取出并发射到下游
• 由上看observeOn可以多次生效

五、背压

• 默认策略判断是否触发背压的因素：
  • 同步场景中，有发射数是否超出响应式拉取值 request 决定
  • 异步场景中，由是否超出缓冲池 queue 的承受能力决定。需要下游的request方法拉取queue的数据
  • observeOn允许我们设置缓冲队列的容量大小
  • 在onSubscribe(Subscription s)回调提供的s可以调用request方法来增加拉取数；如果不重写，默认执行s.request(Long.MAX_VALUE)
• 背压策略
  • ERROR 触发背压直接抛异常 MissingBackpressureException
  • BUFFER: queue无限大，知道OOM
  • DROP: 超载则抛弃之后的数据，不抛异常
  • LATEST：超载后抛弃之后数据，且是专用有一个额外空间保留当前最新一次数据

举例 onBackpressureDrop 方法

public final class FlowableOnBackpressureDrop<T> extends AbstractFlowableWithUpstream<T, T> implements Consumer<T> {
    final Consumer<? super T> onDrop;
  
    @Override
    protected void subscribeActual(Subscriber<? super T> s) {
        this.source.subscribe(new BackpressureDropSubscriber<T>(s, onDrop));
    }
  
    static final class BackpressureDropSubscriber<T> extends AtomicLong implements FlowableSubscriber<T>, Subscription {
      @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            long r = get();
            if (r != 0L) {
                actual.onNext(t);
                BackpressureHelper.produced(this, 1);
            } else {
                try {
                    onDrop.accept(t);
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    cancel();
                    onError(e);
                }
            }
        }
      
      @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(this, n);
            }
        }
    }
}

BackpressureDropSubscriber继承了AtomicLong，实现了Subscriber。只有判断自身不为0时才会向下游发射元素，否则将被抛弃。这个数值的计算在BackpressureHelper中计算

// 每执行一次onNext，当前值减一
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;
        if (update < 0L) {
            RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
            update = 0L;
        }
        if (requested.compareAndSet(current, update)) {
            return update;
        }
    }
}

// request调用时，值会加到当前值上。
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;
        }
    }
}

这样，就实现了发生背压则抛弃新的值

onBackpressureLatest的实现和前者类似，只是多了一个对象来存储最新一次的值：

final AtomicReference<T> current = new AtomicReference<T>();
@Override
public void onNext(T t) {
     current.lazySet(t);
     drain();
}

六、Subject

Subject 继承Observable，实现Observer。具有自攻自受的特性，相当于是一个中间层

一般Observable观察的对象往往是静态的，如一个常量、一个文件。有确定的开头和结尾

Subject是动态的，监听没有被中断的时候，可以有不确定的事件传来

• Observable特性
  • 订阅：如果下游可用，则加入到订阅队列中

@Override
public void subscribeActual(Observer<? super T> t) {
    PublishDisposable<T> ps = new PublishDisposable<T>(t, this);
    t.onSubscribe(ps);
    if (add(ps)) {
        // if cancellation happened while a successful add, the remove() didn't work
        // so we need to do it again
        if (ps.isDisposed()) {
            remove(ps);
        }
    } else {
        Throwable ex = error;
        if (ex != null) {
            t.onError(ex);
        } else {
            t.onComplete();
        }
    }
}

boolean add(PublishDisposable<T> ps) {
    for (;;) {
        PublishDisposable<T>[] a = subscribers.get();
        if (a == TERMINATED) {
            return false;
        }

        int n = a.length;
        @SuppressWarnings("unchecked")
        PublishDisposable<T>[] b = new PublishDisposable[n + 1];
        System.arraycopy(a, 0, b, 0, n);
        b[n] = ps;

        if (subscribers.compareAndSet(a, b)) {
            return true;
        }
    }
}

• Observer 特性
  • 如果本身可用，切存在可用的观察者，则向下传递事件

// 但举例onNext方法
@Override
public void onNext(T t) {
    if (subscribers.get() == TERMINATED) {
        return;
    }
    if (t == null) {
        onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
        return;
    }
    // 遍历下游观察者队列，并逐个调用onNext
    for (PublishDisposable<T> s : subscribers.get()) {
        s.onNext(t);
    }
}

七、性能问题

.map(x -> x + 1) 和 .flatMap(x -> Observable.just(x + 1))有什么区别

• map和flatMap的区别
  • map没有创建新的Observable，flatMap创建了新的Observable，相当于创建了新的流
  • map在上游和下游之间仍是线性的。flatMap已经上升了一个阶，即每一个元素进来，都会变成一个新的source，而下游也会变成一个新的observer

循环的静态Observable和动态的Subject之间的区别

void callback(int item) {
    Observable
        .<Integer>create(emitter -> {
            emitter.onNext(item);
        })
        .map(i -> i + 1)
        .subscribe(i -> Log.d(TAG, "i = " + i));
}

Subject<Integer> mSubject = PublishSubject.create();

void init() {
    mSuject
        .map(i -> i + 1)
        .subscribe(i -> Log.d(TAG, "i = " + i));
}

void callback(int item) {
    mSuject.onNext(item);
}

• 循环创建的事件流，所有操作符都涉及了new对象，如果callback被频繁调用，则会产生十分多的临时对象造成内存抖动

八、Rx扩展

RxBinding

final class ViewClickObservable extends Observable<Object> {
    private final View view;
    ViewClickObservable(View view) {
        this.view = view;
    }
    @Override protected void subscribeActual(Observer<? super Object> observer) {
        if (!checkMainThread(observer)) {
            return;
        }
        //订阅阶段执行的逻辑，创建listener绑定view
        Listener listener = new Listener(view, observer);
        observer.onSubscribe(listener);
        view.setOnClickListener(listener);
    }
    static final class Listener extends MainThreadDisposable implements OnClickListener {
        private final View view;
        private final Observer<? super Object> observer;
        Listener(View view, Observer<? super Object> observer) {
            this.view = view;
            this.observer = observer;
        }
        @Override public void onClick(View v) {
            if (!isDisposed()) {
                //发生点击时传递事件给下游
                observer.onNext(Notification.INSTANCE);
            }
        }
        @Override protected void onDispose() {
            view.setOnClickListener(null);
        }
    }
}

RxPermissions

public <T> ObservableTransformer<T, Boolean> ensure(final String... permissions) {
    return new ObservableTransformer<T, Boolean>() {
        @Override
        public ObservableSource<Boolean> apply(Observable<T> o) {
            return request(o, permissions)
                    // 转换 Observable<Permission> 为 Observable<Boolean>
                    .buffer(permissions.length)
                    .flatMap(new Function<List<Permission>, ObservableSource<Boolean>>() {
                        @Override
                        public ObservableSource<Boolean> apply(List<Permission> permissions) {
                            if (permissions.isEmpty()) {
                                // Occurs during orientation change, when the subject receives onComplete.
                                // In that case we don't want to propagate that empty list to the
                                // subscriber, only the onComplete.
                                return Observable.empty();
                            }
                            // Return true if all permissions are granted.
                            for (Permission p : permissions) {
                                if (!p.granted) {
                                    return Observable.just(false);
                                }
                            }
                            return Observable.just(true);
                        }
                    });
        }
    };
}

注：本文为了节省篇幅，对源码进行了一定程度的删减，仅供参考。

参考文章

• RxJava github
• RxAndroid github
• RxJava2 源码分析
• RxJava2.x 分析源码,理解操作符FlatMap