跟踪[rx]大型图中可观察到的异常到源代码

Question

当您有一个大的Observable图（即使用merge、groupBy、join等多次组合的Observal图），并且抛出异常时，有时很难弄清楚异常的来源。我想知道是否有可能找出源文件中Observable运算符的调用位置。一个例子应该使这一点更加清楚。

例如，给定以下IllegalStateException: Only one subscriber allowed!和堆栈跟踪，我想知道是否有可能在我的源文件中找出operatorMerge、operatorFilter、operatorGroupBy等的行号是否可以通过使用调试器、打印语句或其他方式以某种方式执行此操作？

java.lang.IllegalStateException: Only one subscriber allowed!
        at rx.internal.operators.BufferUntilSubscriber$OnSubscribeAction.call(BufferUntilS
ubscriber.java:124)
        at rx.internal.operators.BufferUntilSubscriber$OnSubscribeAction.call(BufferUntilS
ubscriber.java:81)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable.unsafeSubscribe(Observable.java:7531)
        at rx.internal.operators.OperatorGroupBy$GroupBySubscriber$2.call(OperatorGroupBy.
java:251)
        at rx.internal.operators.OperatorGroupBy$GroupBySubscriber$2.call(OperatorGroupBy.
java:236)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable$1.call(Observable.java:144)
        at rx.Observable$1.call(Observable.java:136)
        at rx.Observable.unsafeSubscribe(Observable.java:7531)
        at rx.internal.operators.OperatorMerge$MergeSubscriber.handleNewSource(OperatorMer
ge.java:215)
        at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:1
85)
        at rx.internal.operators.**OperatorMerge**$MergeSubscriber.onNext(OperatorMerge.java:1
20)
        at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55)
        at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55)
        at rx.internal.operators.SingleDelayedProducer.emit(SingleDelayedProducer.java:80)
        at rx.internal.operators.SingleDelayedProducer.set(SingleDelayedProducer.java:63)
        at rx.internal.operators.OperatorToObservableList$1.onCompleted(OperatorToObservab
leList.java:93)
        at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:44)
        at rx.internal.operators.**OperatorFilter**$1.onCompleted(OperatorFilter.java:42)
        at rx.internal.operators.OperatorTakeUntilPredicate$ParentSubscriber.onNext(Operat
orTakeUntilPredicate.java:54)
        at rx.internal.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:84)
        at rx.internal.operators.**OperatorGroupBy**$GroupBySubscriber$2$2.onNext(OperatorGrou
pBy.java:286)
        at rx.internal.operators.BufferUntilSubscriber.onNext(BufferUntilSubscriber.java:1
81)
        at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
        at rx.internal.operators.OperatorGroupBy$GroupBySubscriber.emitItem(OperatorGroupB
y.java:340)
        at rx.internal.operators.OperatorGroupBy$GroupBySubscriber.onNext(OperatorGroupBy.
java:226)
        at rx.internal.operators.OnSubscribeRefCount$2.onNext(OnSubscribeRefCount.java:124
)
        at rx.internal.operators.OperatorPublish$PublishSubscriber.dispatch(OperatorPublis
h.java:560)
        at rx.internal.operators.**OperatorPublish**$PublishSubscriber.onNext(OperatorPublish.
java:258)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:676
)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:5
86)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:676
)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:5
86)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:676
)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:5
86)
        at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.request(OnSubscr
ibeFromIterable.java:98)
        at rx.Subscriber.setProducer(Subscriber.java:177)
        at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java
:50)
        at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java
:33)
        at rx.Observable.unsafeSubscribe(Observable.java:7531)
        at rx.internal.operators.OperatorMerge$MergeSubscriber.handleNewSource(OperatorMer
ge.java:215)
        at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:1
85)
        at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:1
20)
        at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55)
        at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55)
        at rx.internal.operators.OnSubscribeRefCount$2.onNext(OnSubscribeRefCount.java:124
)
        at rx.internal.operators.OperatorPublish$PublishSubscriber.dispatch(OperatorPublis
h.java:560)
        at rx.internal.operators.OperatorPublish$PublishSubscriber.onNext(OperatorPublish.
java:258)
        at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55)
        at rx.internal.operators.OperatorScan$2.onNext(OperatorScan.java:112)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:676
)
        at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:5
86)
        at rx.internal.operators.OperatorGroupBy$GroupBySubscriber$2$2.onNext(OperatorGrou
pBy.java:286)
        at rx.internal.operators.BufferUntilSubscriber.onNext(BufferUntilSubscriber.java:1
81)
        at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
        at rx.internal.operators.OperatorGroupBy$GroupBySubscriber.emitItem(OperatorGroupB
y.java:340)
        at rx.internal.operators.OperatorGroupBy$GroupBySubscriber.onNext(OperatorGroupBy.
java:226)
        at rx.lang.scala.Subscriber$$anon$3.onNext(Subscriber.scala:198)
...

这个问题的出现本质上是因为Observable的全部目的是在执行时将a）代码与b）解耦。但对于调试程序来说，这是一场噩梦。因此，重复我上面的问题，我想知道是否有可能将每个组成部分追溯到源代码中的原始行。

Answer 1

有一些关于额外调试信息的实验，但整个库的运行速度慢了100倍，因此被放弃。

问题可能出现在你的groupBy之后的flatMap中，你订阅了GroupedObservable，并将其交回了现在无法订阅的flatMap:GroupedObserver只能使用一次。您需要使用publish()或replay()运算符之一，并相应地调整函数逻辑。

Answer 2

一年过去了，我仍然在与之斗争，仍然没有找到追踪处决的好方法。我发现我依赖于在代码中放入print语句来查看发生了什么。这是我可以跟踪发生了什么的唯一方法。

我发现唯一有帮助的是为此创建一个模式，这样我就不必编写doOnNext(x => println(x))来显示每次发生的事情：

  implicit class ObservableTrace[T](o : rx.lang.scala.Observable[T]) {
    import java.time.LocalTime
    def trace(name : String) : rx.lang.scala.Observable[T] = {
      def print(s: String) = println(s"${LocalTime.now} : $name : $s")
      (o doOnNext (x => print("next:" + x))
        doOnSubscribe print("subscribed")
        doOnCompleted print("completed")
        doOnError (e => print("error: " + e))
        doOnUnsubscribe print("unsubscribed")
        )
    }

这使得编辑代码变得很快——只需在几个可观察的对象上编写myobservable.trace("My Observable")，就可以很容易地看到不同的生命周期事件何时发生。