为什么在我的可观察序列中需要第二次onSubscribe（）调用？

Question

需要在我的可观测序列中第二次应用onSubscribe()算子，请参见行：details.add(myApi.getDetails(h.getId()).subscribeOn(Schedulers.io()));。如果未应用onSubscribe()运算符，则会引发NetworkOnMainThreadException。

我的理解是，由于我已经在序列的早期应用了subscribeOn(Schedulers.io())算子，所有未来的订阅应该发生在io调度器上。我的理解有什么问题？这可能是一个改进的beta2问题，因为在下面的示例中，myApi实例是通过Retrofit创建的？

myApi.getHeadlines() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .flatMap(new Func1<Headlines, Observable<HeadlineDetail> { 
     @Override 
     public Observable<HeadlineDetail> call(Headlines headlines) { 
      List<Observable<HeadlineDetail>> details = new ArrayList<>(); 
      for (Headline h : headlines) { 
        details.add(myApi.getDetails(h.getId()).subscribeOn(Schedulers.io())); 
      } 
      return Observable.merge(details); 
     } 
    }) 
    .subscribe(...); 

依赖关系：

• 改造贝塔2.0-β2
• 改造rxjava适配器2.0-β2
• rxjava v1.0.14
• rxandroind V1.0.1

Answer 1

subscribeOn设置observable开始的线程，但是observeOn影响用于下游操作的线程。他们正在“观察”最初的可观察物。你从后台线程开始，但切换到主线程。尝试将您的observeOn移到更晚的链中。

有关更多详细信息，请参阅observeOn上的文档。

myApi.getHeadlines() 
    .subscribeOn(Schedulers.io()) 
    .flatMap(new Func1<Headlines, Observable<HeadlineDetail> { 
     @Override 
     public Observable<HeadlineDetail> call(Headlines headlines) { 
      List<Observable<HeadlineDetail>> details = new ArrayList<>(); 
      for (Headline h : headlines) { 
        details.add(myApi.getDetails(h.getId())); 
      } 
      return Observable.merge(details); 
     } 
    }) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(...);