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がある

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

します単純なPublishSubject。 RxJava 1.1.0と一緒にRxAndroid 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 

私はeventbusとしてSubjectを使用します上記のように、非常に頻繁に。そして、私はそのような問題はなかった

P.S. onDetachedFromWindow()では、サブスクリプションがサブスクリプションであるかどうかを確認する方が良いでしょう。私は、このメソッドはメインスレッドで呼び出されることを知っており、このSubscriptionへの同時アクセスは不可能ですが、私はそれが良いスタイルだと思う：答えのための

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