使用未收到的消息调试RxJava问题的最佳方法是什么

我有一个Android应用程序,其中有多个A类观察者订阅了几个类型为B的Observable.订阅在IO Scheduler中完成,并在Android主线程上观察.

问题是我在一些工作之后随机发现 B发出的一条消息从未在A中收到过,经过几个小时的重新调整后我无法找到原因.

问题发生时的相关代码:

"NEXT1"和"NEXT2"被打印但"收到","错误","完成"不是.

            //The subscription
            B.getMessate()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(A);

            //B
            Observable getMessage() {
                 return Observable.create(new Observable.OnSubscribe() {    
                      public void call(Subscriber subscriber) {
                         ...
                         subscriber.onNext(msg)
                         println("NEXT1")
                      }
                 }).doOnNext({ (o) -> println("NEXT2")});
            }



            //A 
            onNext(msg) {
                  //Never called when problem happens
                  println("RECEIVED")
            }
            onError(msg) {
                  //Never called when problem happens
                  println("ERROR")
            }
            onError(msg) {
                  //Never called when problem happens
                  println("COMPLETED")
            }

任何人都有一些线索？或任何调试建议？

我检查了什么:

我暂停了应用程序并检查了所有线程以查看是否已锁定.并且所有工作线程都被停放,主线程正在等待android消息队列中的消息.

观察员从不打电话取消订阅()

lujop.. 22

到现在为止我无法重现这个问题,但我发现RxJavaDebug是一个非常好的调试工具.

它的用法很简单:将库添加为依赖项,并在应用程序启动时注册一个监听器:

  RxJavaPlugins.getInstance().registerObservableExecutionHook(new DebugHook(new DebugNotificationListener() {
      public Object onNext(DebugNotification n) {
          Log.v(TAG, "onNext on " + n);
          return super.onNext(n);
      }


        public Object start(DebugNotification n) {
            Log.v(TAG, "start on " + n);
            return super.start(n);
        }


        public void complete(Object context) {
            Log.v(TAG, "complete on " + context);
        }

        public void error(Object context, Throwable e) {
            Log.e(TAG, "error on " + context);
        }
  }));

这将在可观察者和运营者之间记录消息.