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

我有一个 Android 应用程序，其中包含多个A型观察者订阅多个B 类可观测量。订阅是在IO Scheduler中完成的，观察是在Android主线程上完成的。

我遇到的问题是随机地经过一些工作后，A 从未收到 B 发出的一条消息，并且经过几个小时的调试后我找不到原因。

相关代码当问题发生时:

打印“NEXT1”和“NEXT2”，但不打印“RECEIVED”、“ERROR”、COMPLETED。

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

            //B
            Observable<msg> getMessage() {
                 return Observable.create(new Observable.OnSubscribe<msg>() {    
                      public void call(Subscriber<? super msg> 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 消息队列中等待消息。
• 观察者永远不会调用 unsubscribe()

到目前为止我无法重现该问题，但我发现RxJava调试 https://github.com/ReactiveX/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);
        }
  }));

这将在消息在可观察量和运算符之间传递时记录消息。