我正在尝试将“第三方”发布者包装到 RxJava 中。在订阅期间，我正在准备发布者，这可能会导致异常（当发布者不可用时）。
异常应该怎么办？
1) 从订阅方法中抛出？
2) 将其推送到观察者的OnError 方法中？
3) 别的？
我正在考虑推送到OnError，那么如何实现订阅呢？如果OnError应该是订阅中的最后一个事件，我应该认为订阅发生了吗？什么返回Subscription？只是NOP lambda？
【问题讨论】：
标签： java rx-java reactivex subscriber
您所说的“发布者”和“包装”有点不清楚。
例如，有一个返回 Observable 的 API 调用，但调用本身可能会引发未经检查的异常。一种简单的方法是使用defer，它将捕获该异常并将其转发给订阅者：
Observable.defer(() -> api.unreliableMethod())...
如果你的意思是你有一个 Reactive-Streams Publisher 和 RxJava 2，你可以像上面一样简单地调用 Flowable.fromPublisher 或 Flowable.defer(() -> publisher)。
否则，RxJavaReactiveStreams 让您将Publisher 转换为Observable（并在需要时延迟（）捕获异常的过程）：
Observable.defer(() ->
RxJavaReactiveStreams.toObservable(api.unreliablePublisher()))...
【讨论】：
我正在考虑推送到OnError，那么如何实现订阅呢？如果 OnError 应该是订阅中的最后一个事件，我应该认为订阅发生还是不发生？作为订阅返回什么？只是 NOP lambda？
这实际上是 RxJava 本身通常在每个运算符的“构造函数”中处理异常的方式。
如果您使用的是 RxJava2，那么您需要记住另外一条规则。那就是Subscriber.onSubcribe()必须总是在onError和onComplete之前被调用，所以即使它立即抛出异常，onSubscribe和onError都必须被调用。
【讨论】：