在Android上调试时,RxJava缓存线程中的InterruptedException

有时当我调试我的应用程序时,我在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)

我有一个自定义视图,我在其中订阅我的observable,如下所示：

@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.

有谁知道为什么会这样？

解决方法

我不确定,为什么会发生,但尝试这样做：

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.我从来没有遇到过这样的问题.

附：
在onDetachedFromWindow()中,如果你检查订阅是否取消订阅会更好.我知道这个方法在主线程中调用,并且对这个Subscription的并发访问是不可能的,但我认为它的风格很好：

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

在Android上调试时,RxJava缓存线程中的InterruptedException

解决方法

猜你在找的Android相关文章