如何在 RxJava 中的 Observable 中处理 map() 中的异常答案

【问题标题】：How do I handle exceptions in map() in an Observable in RxJava如何在 RxJava 中的 Observable 中处理 map() 中的异常
【发布时间】：2016-02-06 23:13:15
【问题描述】：

我想这样做：

Observable.just(bitmap)
            .map(new Func1<Bitmap, File>() {
                @Override
                public File call(Bitmap photoBitmap) {

                    //File creation throws IOException, 
                    //I just want it to hit the onError() inside subscribe()

                    File photoFile = new File(App.getAppContext().getCacheDir(), "userprofilepic_temp.jpg");
                    if(photoFile.isFile()) {//delete the file first if it exists otherwise the new file won't be created
                        photoFile.delete();
                    }
                    photoFile.createNewFile(); //saves the file in the cache dir

                    FileOutputStream fos = new FileOutputStream(photoFile);
                    photoBitmap.compress(Bitmap.CompressFormat.JPEG, 90, fos);//jpeg format
                    fos.close();

                    return photoFile;

                }
            })
            .subscribe(//continue implementation...);

基本上在call()方法中，它可以抛出异常。如何让观察者在onError() 中处理它。或者这不是正确的思考方式？

【问题讨论】：

请注意，在 RxJava 2 中，map 等运算符允许从 lambda 中抛出已检查的异常。这确实是 RxJava 1 中的一个设计缺陷，因为如果不包装为 RuntimeException，则无法在映射 lambda 的 'onError` 中传播引发的确切错误。

标签： java android rx-java rx-android

【解决方案1】：

在 1.0.15 中，有 fromCallable 工厂方法，让您可以为每个订阅者运行一个 Callable 实例，您也可以在其中抛出已检查的异常：

Observable.fromCallable(() -> {      
    File photoFile = new File(App.getAppContext().getCacheDir(),
        "userprofilepic_temp.jpg");
    if (photoFile.isFile()) {
       //delete the file if it exists otherwise the new file won't be created
        photoFile.delete();
    }
    photoFile.createNewFile(); //saves the file in the cache dir

    FileOutputStream fos = new FileOutputStream(photoFile);
    photoBitmap.compress(Bitmap.CompressFormat.JPEG, 90, fos);//jpeg format
    fos.close();

    return photoFile;
})
.subscribe(...)

编辑：

source.flatMap(v -> {
    try {
        //...
        return Observable.just(result);
    } catch (Exception e) {
        return Observable.error(e);
    }
})
.subscribe(...);

【讨论】：

我想这适用于这种情况，但是如果我想将两个依赖于第一个地图输出的地图链接在一起怎么办。使用Func1()我可以得到返回类型和参数。
你必须使用 flatMap，try-catch 并返回 just() 或 error()。
像冠军一样工作！！

【解决方案2】：

我不知道当这个问题第一次被问和回答时的情况如何，但 RxJava 目前包含一个用于这个确切目的的辅助方法： Exceptions.propagate(Throwable t)

RxJava Javadoc

直接抛出 RuntimeException 和 Error 或将任何其他异常类型包装到 RuntimeException 中的便捷方法。

【讨论】：

@Sree 哇，我实际上没有发现这一点，我被自定义异常类蒙蔽了双眼。现在我仔细观察，我可以看到这里实际上有另一个使用相同方法的答案。

【解决方案3】：

rx 将始终捕获错误，即使这是 RuntimeException。所以你可以在 catch 块中抛出某种运行时异常。实际上它应该是这样工作的。

 Observable.just(bitmap)
                .map(b -> {
                    try {
                        // do some work which throws IOException
                        throw new IOException("something went wrong");
                    } catch (IOException e) {
                        throw new RXIOException(e);
                        // Or you can use 
                        throw Exceptions.propagate(e);
                        // This helper method will wrap your exception with runtime one
                    }
                }).subscribe(o -> {
                    // do something here
                }, exception -> exception.printStackTrace());

public static class RXIOException extends RuntimeException {
        public RXIOException(IOException throwable) {
            super(throwable);
        }
}

【讨论】：

当我尝试这样做时，它要求我用 try-catch 包围异常
你必须抛出 RuntimeException。 JVM 不会要求你用 try catch 包围它。就是这个想法
我让call() 抛出一个 IOException 但它说被覆盖的方法不会抛出 IOException
来吧。 IOException 不是运行时异常。再读一遍我的答案。
啊，对不起。完全有道理。你能看看上面使用Observable.error 的@akarnokd 的答案吗？你会推荐你的方法吗？

【解决方案4】：

刚刚创建了帮助类来将此样板提取到另一个地方：

public class RxRethrow {
    public static <T, R> Func1<T, R> rethrow(Func1R<T, R> catchedFunc) {
        return t -> {
            try {
                return catchedFunc.call(t);
            } catch (Exception e) {
                throw Exceptions.propagate(e);
            }
        };
    }

    public interface Func1R<T, R> extends Function {
        R call(T t) throws Exception;
    }
}

你可以这样称呼它：

.map(RxRethrow.rethrow(products -> mapper.writer(schema).writeValueAsString(products)))

【讨论】：