在这个简单的 RxJava 异步场景中，我真的需要进行异常处理吗？

Question

我对 RxJava 非常陌生，虽然我熟悉流，并且对 Javascript 承诺有些熟悉。我正在处理一些使用 RxJava 的现有代码，以及一些人对这些代码所做的一些 cmets。我想了解更多这方面的背景知识，同时我会继续吸收文档。

有问题的块是这样的（一些名称已更改）：

public ShippingMethodHolder callClientsAsync(ShippingMethodContext shippingContext) {
    Single<ShippingMethodResponse> productOneResponseEntity = Single.<ShippingMethodResponse>create(source -> {
        source.onSuccess(getProductOneowShippingMethodResponse(shippingContext));
    }).subscribeOn(Schedulers.io());

    Single<ShippingMethodResponse> productTwoResponseEntity = Single.<ShippingMethodResponse>create(source -> {
        source.onSuccess(getProductTwoShippingMethodResponse(shippingContext));
    }).subscribeOn(Schedulers.io());

    Single<ShippingMethodHolder> singleProductCartResponseHolder = Single.zip(productOneResponseEntity, productTwoResponseEntity,
            (dtvResponse, productTwoResponse) -> {
                return new ShippingMethodHolder(dtvResponse, productTwoResponse);
            });
    return singleProductCartResponseHolder.blockingGet();
}

对此代码有更多了解的人对这段代码的评论基本上是说这缺少 RxJava 异常处理“并将导致流的阻塞或崩溃”。我想这是指两个异步调用有“onSuccess()”调用，但没有“onError()”调用。

但是，这对我来说似乎很奇怪。 "onSuccess()" 被调用的范围不是业务逻辑成功或失败，而是似乎在 RxJava 尝试进行异步调用。

对于从 RxJava 的角度来看这是否真的是一个问题，我可以提出一些建议。

Answer 1

create 主要用于将异步源与反应性世界联系起来，但您的代码似乎阻塞地调用某些东西只是为了表明它的价值。为此，fromCallable 更合适，可以更好地向读者传达意图：

Single<ShippingMethodResponse> productOneResponseEntity = 
    Single.<ShippingMethodResponse>fromCallable(() -> 
        getProductOneowShippingMethodResponse(shippingContext)
    )
    .subscribeOn(Schedulers.io());

根据您的应用程序类型，可能不希望阻塞等待结果，尤其是从 UI 线程调用该方法时。您可以返回压缩后的 Single 并继续编写，直到可以发出最终的 subscribe()。

对此代码有更多了解的人对这段代码的评论基本上说这是缺少 RxJava 异常处理“并且会导致流的阻塞或崩溃”

原来的create 和fromCallable 会捕捉到你的异常，并尝试向消费者发出信号。在这种情况下，blockingGet 将在调用者线程上重新抛出一个源异常，而另一个（如果有）将被路由到全局 RxJavaPlugins.onError 处理程序。他们可能意味着您的方法的调用者通常不希望它抛出，因此他们可能会忽略它周围的 try-catch 并在运行时严重失败。解决它实际上取决于您在应用程序中打算使用哪种类型或错误管理。