关于RxJava可能被忽略的X个知识点

featured-rx-android.png

一 操作符之痛

在RxJava中，我们使用最爽的莫过于对一个事件流使用各种操作符，使其按照我们的指令执行各种操作，读过源码的同学对lift比较熟悉，它是大部分操作符实现的基础，但是一个lift操作可能会产生多个多余的对象。我们以最简单的Map为例，由于使用了lift操作，我们至少生成了三个多余的类：Observable OnSubscribe 和 一个OperatorMap，在某些特殊的场景下面，比如 flatMap操作符，如果我们在内层Observable中使用了各种操作符，那么会产生很多短生命周期的对象，这意味着更多的内存垃圾和GC抖动。

所以我们在使用操作符的时候，需要留意在实现相同转换效果的情况下，是否可以对操作符组合做一定的优化。

Observable... .map().filter()     VS   Observable... .filter().map()

显然后面的表达方式更高效，因为被过滤掉的元素就不需要进行map操作了。

 Observable...flatMap(return observable.map();)   VS Observable...flatMap(return observable;).map()

第一种表达方式中相当于对内部创建的Observable做一些转换操作，我们知道大部分转换操作都会重新生成新的Observable，所以在内部Observable中使用各种转换会产生较多的中间对象。

 Observable...flatMap(returnobservable.subscribeOn(Schedulers.io());)
                  .subscribeOn(Schedulers.io())  //多余
                  .observeOn(AndroidSchedulers.mainThread())
                  .map()

多余的线程跳跃。

二 错误处理

众所周知，RxJava一大优势就是能够将错误处理集中起来到最后订阅者的onError中统一处理，那么如果我们在onNext、onCompleted或者onError中又抛出了异常，那么错误会被catch住还是直接crash呢？我们来看看源码吧：

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
    try {
        //事件源逻辑
        hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
        return hook.onSubscribeReturn(subscriber);
    } catch (Throwable e) {  //(1)
        Exceptions.throwIfFatal(e);//(2)
        try {
            subscriber.onError(hook.onSubscribeError(e));//(3)
        } catch (Throwable e2) {//(4)
            Exceptions.throwIfFatal(e2);
            RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
            hook.onSubscribeError(r);
            throw r;
        }
    }
}

我们可以看到事件源也就是我们调用onNext或者onCompleted的地方一旦发生异常，那么就会走到(1)被Catch住，这里至少说明你在onNext或者onCompleted中所产生的异常都会被兜住，在(3)传递给onError，当然SafeSubscriber做了点小魔法，不会让onCompleted产生的错误传递给onError，因为按照Rx规范，一个流只可能执行onCompleted和onError中之一作为结束。

Rx规范 的保证大部分是通过SafeSubscriber这个类来实现的，我们通过普通的Observable.subscribe()进行订阅时，会自动包装成SafeSubscriber

如果我们在onError中处理产生了错误，那么会走到(4),此时RxJava会直接把这个错误抛出，导致crash。

上面的(2)位置的Exceptions.throwIfFatal是一个Utils方法，它会查看当前异常类型，如果是毁灭级别的异常(如 OOM)，RxJava会直接抛出来而不是交给应用逻辑去处理。

我们前面讨论的大部分都是非受检的异常，如果我们需要在响应流中调用抛出受检异常的方法，这就会比较尴尬，因为在Rx 1.x版本中绝大部分函数签名都不允许抛出异常，比如我们有如下场景：

//做一个转化的已有逻辑，需要抛出来IOException
String transform(String s) throws IOException 
//在某一个数据流中需要如下处理：
  Observable.just("Nice!")  
          .map(input -> {
                  try {
                    return transform(input);
                   } catch (Throwable t) {
                     throw Exceptions.propagate(t);
                   }
                }
      );

大部分人会推荐你使用上面的做法，使用Exceptions.propagate将一个受检异常包装起来然后抛出，然而这样合理吗？

在订阅者收到这个Throwable之后，他需要自己收到去掉一层包围才能看到真实的异常，显然订阅者需要知道整个流中是怎么包装这个异常的，看起来不是太cool，那试试下面的方式呢？

Observable.just("Nice!")  
          .flatMap(input -> {
                  try {
                    return Observable.just(transform(input));
                   } catch (Throwable t) {
                     return Observable.error(t);
                   }
                }
      );

完美。

三 使用背压

反压或者背压是RxJava中提供的一种订阅者和数据源进行反向沟通的渠道，通常大部分文章都会说它是为了解决数据源发送过快而接受者来不及处理收到的消息；从这个层面来说，我一直认为反压貌似在Android开发中没有什么使用场景，因为我们毕竟不会涉及到高并发，应该不会出现需要使用反压的情况吧？我相信大部分人应该都有这种想法。直到我遇到下面的场景的时候，对反压的认识就更加准确了。

有个页面是由5个模块组成，每个模块都一个关联的数据请求，我们要求进入页面之后，从上往下加载按序加载，先加载第一个，然后等第一个加载完了然后加载第二个，如果某一个失败，那么停止加载剩下的模块。那么这个模型可以用下面的伪代码表示：

static Observable[] dataRequest = new Observable[5];
static void go() {
    Observable.from(dataRequest)
            .subscribe(new Subscriber<Observable>() {

                @Override
                public void onStart() {
                    super.onStart();
                    request(1);
                }

                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(Observable observable) {
                    request(1);
                    //dorequest
                }
            });
}

这个看起来不太简洁，而且在dorequest处其实流已经被打断了，我们可以使用万能的flatMap来解决：

Observable.from(dataRequest)
            .flatMap(new Func1<Observable, Observable<?>>() {
                @Override
                public Observable<?> call(Observable observable) {
                    return observable;
                }
            },1);

注意，flatMap是支持背压的，所以我们给了背压参数为1。

四 串行调用

首先我们来个反例来说明一下问题：

    Subject<String,String > publishSubject = PublishSubject.create();
    publishSubject.subscribe(new Subscriber<String>() {

        private String mString = "";


        @Override
        public void onCompleted() {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(String s) {
            System.out.println(Thread.currentThread().getName()+" doing onNext");
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+" out onNext");
        }
    });


    new Thread(new Runnable() {
        @Override
        public void run() {
            publishSubject.onNext("thread1");
        }
    }).start();

    new Thread(new Runnable() {
        @Override
        public void run() {
            publishSubject.onNext("thead2");
        }
    }).start();

  //输出
  Thread-0 doing onNext
  Thread-1 doing onNext
  Thread-0 out onNext
  Thread-1 out onNext

我们看到了什么？订阅者的onNext被同时调用了，这可跟事件流定义可不一样，我们印象里的事件流都是一个接一个串行的，现在订阅者居然同时收到了两个事件。同时收到事件会有什么问题呢？很明显，大部分的订阅者都是被设计成在同步模式下工作的，如果同时有多个线程调用onNext方法，那就会破坏订阅者内部的状态。

这就是Rx规范中的串行访问，作为Rx规范第一条，它的重要性不言而喻，但是有可能你使用了很久的RxJava都没有注意到，下面是原文。

Observables must issue notifications to observers serially (not in parallel). They may issue these notifications from different threads, but there must be a formal happens-before relationship between the notifications.

在Rx中有一些操作是做线程合流(join)的，比如merge flatMap等等，想象一下，merge的多个Observable会运行在不同的线程上面，如果merge操作符不做任何特殊处理，merge操作符的订阅者状态就会被破坏。如果去保证onNext是被串行调用的呢？一个简单的想法肯定首先冒出来，我给订阅者的onNext加把锁做同步不就可以了？比如使用synchronized来描述onNext,但是你要知道，在Rx中数据表面上是从上游往下游流动的，但是有很多数据沟通是从下游往上游走的(比如上面说的反压)，如果上游往下游发数据的过程中，下游也需要往上游沟通，那么中间的订阅者就很有可能发生死锁。所以在RxJava中采用串行发送器来实现这个功能：

class EmitterLoopSerializer {
  boolean emitting;
  boolean missed;
  public void emit() {
    synchronized (this) {         
        if (emitting) {
            //暂存事件，由当前正在发送的线程事件
            missed = true;         
            return;
        }
        emitting = true;            
    }
    for (;;) {
        //尽可能多的发送事件
        synchronized (this) {  
            if (!missed) {       
                emitting = false;
                return;
            }
            missed = false;    
        }
    }
}
}

这里其实是把锁访问代理到了EmitterLoopSerializer，而不是订阅者上面，这样就可以很好的避免死锁的问题。