在Android上進行調試時RxJava緩存線程中的InterruptedException

Question

有時當我調試我的應用程序時，我遇到了RxCachedThreadScheduler-1中的InterruptedException。這裏的痕跡：

Fatal Exception: java.lang.InterruptedException 
     at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:1991) 
     at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2025) 
     at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1048) 
     at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:776) 
     at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1035) 

我有我訂閱我觀察到這樣的自定義視圖：

@Override 
protected void onAttachedToWindow() { 
    super.onAttachedToWindow(); 

    sub = FollowHandler.getInstance().getObservable() 
      .filter(new Func1<FollowEvent, Boolean>() { 
       @Override 
       public Boolean call(FollowEvent followEvent) { 
        if(followEvent == null || followEvent.user == null 
          || user == null) 
         return false; 

        return followEvent.user.id == user.id; 
       } 
      }) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Observer<FollowEvent>() { 
       @Override 
       public void onCompleted() {} 

       @Override 
       public void onError(Throwable e) {} 

       @Override 
       public void onNext(FollowEvent followEvent) { 
        reactToThisNiceEvent(followEvent); 
       } 
      }); 
} 

@Override 
protected void onDetachedFromWindow() { 
    super.onDetachedFromWindow(); 

    if(sub != null) 
     sub.unsubscribe(); 
} 

這裏的觀察到：

其中

eventSubject.asObservable() 
     .observeOn(Schedulers.io()) 
     .doOnNext(new Action1<FollowEvent>() { 
      @Override 
      public void call(FollowEvent followEvent) { 
       if(followEvent != null) 
        doSomethingNice(followEvent); 
      } 
     }) 
     .share(); 

eventSubject是簡單的PublishSubject。 我使用RxAndroid 1.1.0和RxJava 1.1.0。

有誰知道爲什麼會發生這種情況？

Answer 1

我不知道，爲什麼會發生，但儘量要做到這一點：

sub = FollowHandler.getInstance().getObservable() 
      .filter(...) 
      .subscribeOn(Schedulers.io()) // <<<<<<<<<< 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(...); 

另外，我覺得你不需要share()：

eventSubject.asObservable() 
     .doOnNext(...) 
     .subscribeOn(Schedulers.io()) // <<<<< subscribeOn instead of observeOn, but actually, you don't need it here... 
     .share();  // <<<<< remove it 

我用Subject作爲eventbus就像我上面所描述的那樣。我從來沒有這樣的問題。

P.S. 在onDetachedFromWindow()會更好，如果你會檢查訂閱取消訂閱或不。我知道，這種方法被稱爲在主線程和併發訪問此Subscription是不可能的，但我認爲這是很好的風格：

if(sub != null && !sub.isUnsubscribed()) 
     sub.unsubscribe();