RxJava合并Observable——merge和mergeDelayError操作符

http://blog.****.net/shuxiangxingkong/article/details/52516018

 

 一，merge操作符可以合并多个Observable的输出，它们的数据可能会交错发射（concat可以保持顺序）。如果某个原始Observable出现onError，merge后的Observable就会onError

 二，mergeDelayError在合并和交错输出上和merge一样，但是mergeDelayError的原始observable出现onError时，错误通知会被保留，直到所有数据发射完毕后才执行onError。如果有多个原始observable出现了onError，这些onError通知会被合并成一个CompositeException ，保留在它的 List<Throwable> exceptions异常列表里。如果只有一个原始observable出现了onError，则会直接使用这个onError通知，而不会生成CompositeException。

见MergeSubscriber里的代码

 

[java] view plain copy

1. private void reportError() {  
2.     ArrayList list = new ArrayList(this.errors);  
3.     if(list.size() == 1) {  
4.         this.child.onError((Throwable)list.get(0));  
5.     } else {  
6.         this.child.onError(new CompositeException(list));  
7.     }  
8.   
9. }  

 

这样如果想对原始observable的异常原因有针对性第处理，就要重写最后subscriber的onError方法，区分两种情况。

 

[java] view plain copy

1. <pre name="code" class="java">                      public void onError(Throwable e) {  
2.                           if (e instanceof CompositeException) {  
3.                               CompositeException compositeException = (CompositeException) e;  
4.                               e = compositeException.getExceptions().get(0);  
5.                           }  
6.                           super.onError(e);  
7.                       }  

 

 

 

 

 

三，mergeDelayError有个坑

 

[java] view plain copy

1. Observable.mergeDelayError(Observable.error(new Exception("错误")),Observable.just("数据"))  
2.            .subscribeOn(Schedulers.io())  
3.           .observeOn(AndroidSchedulers.mainThread())  
4.            .subscribe(new Subscriber<String>() {  

 

 

 

发现这样先mergeDelayError再指定线程的话，mergeDelayError没有起到延迟通知onError的作用，第一个observable出现错误的时候，整个合并的observable也onError了，第二个observable无法输出。

但是如果改成每个observable单独subscribeOn和observeOn，然后再mergeDelayError，就正常进行了。

 

 

[java] view plain copy

1. Observable.mergeDelayError(Observable.error(new Exception("错误"))  
2.                                      .subscribeOn(Schedulers.io())  
3.                                      .observeOn(AndroidSchedulers.mainThread()),  
4.          Observable.just("数据").subscribeOn(Schedulers.io())  
5.                          .observeOn(AndroidSchedulers.mainThread()))  
6.            .subscribe(new Subscriber<String>() {