你什么时候在RxJava中使用map vs flatMap？

Question

你什么时候在RxJava中使用map vs flatMap？

比方说,我们想将包含JSON的文件映射到包含JSON的字符串中 -

使用map,我们必须以某种方式处理Exception.但是如何？:

Observable.from(jsonFile).map(new Func1<File, String>() {
    @Override public String call(File file) {
        try {
            return new Gson().toJson(new FileReader(file), Object.class);
        } catch (FileNotFoundException e) {
            // So Exception. What to do ?
        }
        return null; // Not good :(
    }
});

使用flatMap,它更加冗长,但我们可以将问题转发到Observables链中,如果我们选择其他地方甚至重试,则可以处理错误:

Observable.from(jsonFile).flatMap(new Func1<File, Observable<String>>() {
    @Override public Observable<String> call(final File file) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override public void call(Subscriber<? super String> subscriber) {
                try {
                    String json = new Gson().toJson(new FileReader(file), Object.class);

                    subscriber.onNext(json);
                    subscriber.onCompleted();
                } catch (FileNotFoundException e) {
                    subscriber.onError(e);
                }
            }
        });
    }
});

我喜欢地图的简单性,但是flatmap的错误处理(不是冗长).我没有看到任何关于这种浮动的最佳实践,我很好奇这是如何在实践中使用的.

Answer 1

map将一个事件转换为另一个 flatMap将一个事件转换为零个或多个事件.(这取自IntroToRx)

由于您希望将json转换为对象,因此使用map应该足够了.

处理FileNotFoundException是另一个问题(使用map或flatmap无法解决此问题).

要解决您的异常问题,只需使用Non checked异常抛出它:RX将为您调用onError处理程序.

Observable.from(jsonFile).map(new Func1<File, String>() {
    @Override public String call(File file) {
        try {
            return new Gson().toJson(new FileReader(file), Object.class);
        } catch (FileNotFoundException e) {
            // this exception is a part of rx-java
            throw OnErrorThrowable.addValueAsLastCause(e, file);
        }
    }
});

与flatmap完全相同的版本:

Observable.from(jsonFile).flatMap(new Func1<File, Observable<String>>() {
    @Override public Observable<String> call(File file) {
        try {
            return Observable.just(new Gson().toJson(new FileReader(file), Object.class));
        } catch (FileNotFoundException e) {
            // this static method is a part of rx-java. It will return an exception which is associated to the value.
            throw OnErrorThrowable.addValueAsLastCause(e, file);
            // alternatively, you can return Obersable.empty(); instead of throwing exception
        }
    }
});

您也可以在flatMap版本中返回一个新的Observable,它只是一个错误.

Observable.from(jsonFile).flatMap(new Func1<File, Observable<String>>() {
    @Override public Observable<String> call(File file) {
        try {
            return Observable.just(new Gson().toJson(new FileReader(file), Object.class));
        } catch (FileNotFoundException e) {
            return Observable.error(OnErrorThrowable.addValueAsLastCause(e, file));
        }
    }
});

Answer 2

FlatMap的行为与map非常相似,不同之处在于它所应用的函数本身会返回一个observable,因此它非常适合映射异步操作.

在实际意义上,Map应用函数只是对链式响应进行转换(不返回Observable); 而FlatMap函数应用返回一个Observable<T>,这就是为什么如果你计划在方法内进行异步调用,建议使用FlatMap的原因.

摘要:

• Map返回T类型的对象
• FlatMap返回一个Observable.

这里可以看到一个明显的例子:http://blog.couchbase.com/why-couchbase-chose-rxjava-new-java-sdk.

Couchbase Java 2.X客户端使用Rx以方便的方式提供异步调用.由于它使用Rx,因此它具有方法map和FlatMap,其文档中的解释可能有助于理解一般概念.

要处理错误,请覆盖您的susbcriber上的onError.

Subscriber<String> mySubscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) { System.out.println(s); }

    @Override
    public void onCompleted() { }

    @Override
    public void onError(Throwable e) { }
};

查看此文档可能会有所帮助:http://blog.danlew.net/2014/09/15/grokking-rxjava-part-1/

有关如何使用RX管理错误的良好来源,请访问:https://gist.github.com/daschl/db9fcc9d2b932115b679

Answer 3

在您的情况下,您需要地图,因为只有1个输入和1个输出.

map - 提供的函数只接受一个项目并返回一个项目,该项目将进一步向下发射(仅一次).

flatMap - 提供的函数接受一个项目,然后返回一个"Observable",意味着新的"Observable"的每个项目将进一步单独发出.

可能是代码将为您清除:

Observable.just("item1").map( str -> {
    System.out.println("inside the map " + str);
    return str;
}).subscribe(System.out::println);

Observable.just("item2").flatMap( str -> {
    System.out.println("inside the flatMap " + str);
    return Observable.just(str + "+", str + "++" , str + "+++");
}).subscribe(System.out::println);

输出:

inside the map item1
item1
inside the flatMap item2
item2+
item2++
item2+++

Answer 4

我想到的方式是你flatMap在想要放入的函数map()返回时使用Observable.在这种情况下,您可能仍会尝试使用,map()但这将是不切实际的.让我试着解释一下原因.

如果在这种情况下你决定坚持下去map,你会得到一个Observable<Observable<Something>>.例如,在你的情况下,如果我们使用了一个虚构的RxGson库,它Observable<String>从它的toJson()方法返回一个(而不是简单地返回一个String),它将如下所示:

Observable.from(jsonFile).map(new Func1<File, Observable<String>>() {
    @Override public Observable<String>> call(File file) {
        return new RxGson().toJson(new FileReader(file), Object.class);
    }
}); // you get Observable<Observable<String>> here

在这一点上,subscribe()对这样一个观察者来说是非常棘手的.在它里面你会得到一个Observable<String>你需要再次subscribe()获得价值的东西.这看起来不实用或不好看.

因此,为了使它有用,一个想法是"展平"这个可观察的可观察对象(你可能会开始看到_flat_Map来自何处).RxJava提供了几种方法来展平observable,为了简单起见,我们假设merge是我们想要的.合并基本上会带来一堆可观察的,并且只要它们中的任何一个发出就会发出.(很多人会认为转换是一个更好的默认值.但是如果你只发出一个值,那么无论如何都无关紧要.)

所以修改我们以前的代码片段,我们会得到:

Observable.from(jsonFile).map(new Func1<File, Observable<String>>() {
    @Override public Observable<String>> call(File file) {
        return new RxGson().toJson(new FileReader(file), Object.class);
    }
}).merge(); // you get Observable<String> here

这是非常有用的,因为订阅(或映射,或过滤,或......)您只需获得String价值.(另外,请注意,merge()RxJava中不存在这样的变体,但是如果你理解了merge的想法,那么我希望你也能理解它是如何工作的.)

所以基本上因为merge()当它成功map()返回一个observable并且你不必一遍又一遍地输入它时,它应该只是有用,它flatMap()被创建为一个简写.它像正常map()情况一样应用映射函数,但稍后它不会发出返回值,而是"展平"(或合并)它们.

这是一般用例.它在代码库中最有用,它使用Rx整个地方,并且你有许多方法返回observables,你想要与其他返回observables的方法链接.

在您的用例中,它恰好也很有用,因为map()只能将发出的一个值转换onNext()为另一个值onNext().但它无法将其转换为多个值,根本没有值或错误.正如akarnokd在他的回答中所写的那样(并且请注意,他比我更聪明,可能一般而言,但至少在涉及到RxJava时)你不应该从你的问题中抛出异常map().所以你可以使用flatMap()和

return Observable.just(value);

当一切顺利的时候,但是

return Observable.error(exception);

什么东西都失败了
有关完整代码段,请参阅他的答案:https://stackoverflow.com/a/30330772/1402641

Answer 5

问题是你什么时候在RxJava中使用map vs flatMap？.我认为一个简单的演示更具体.

如果要将发出的项目转换为其他类型,在将文件转换为String的情况下,map和flatMap都可以工作.但我更喜欢地图操作员,因为它更清楚.

但是在某些地方,flatMap可以做神奇的工作而map不能.例如,我想获得用户的信息,但我必须在用户登录时首先获取他的ID.显然我需要两个请求,它们是有序的.

让我们开始.

Observable<LoginResponse> login(String email, String password);

Observable<UserInfo> fetchUserInfo(String userId);

这里有两个方法,一个用于返回登录Response,另一个用于获取用户信息.

login(email, password)
        .flatMap(response ->
                fetchUserInfo(response.id))
        .subscribe(userInfo -> {
            // get user info and you update ui now
        });

如您所见,在函数flatMap中应用,首先我获取用户id Response然后获取用户信息.当两个请求完成后,我们可以完成更新UI或将数据保存到数据库中的工作.

但是,如果你使用map你不能写这么好的代码.总之,flatMap可以帮助我们序列化请求.

Answer 6

下面是一个简单的拇指规则,我用帮我决定的时候使用flatMap()过map()的的Rx的Observable.

一旦你决定采用map转换,你就会编写转换代码来返回一些Object吗？

如果您转换的最终结果是:

• 一个不可观察的对象然后你只使用map().并将map()该对象包装在Observable中并发出它.

• 一个Observable对象,然后你就用了flatMap().并flatMap()解开Observable,选择返回的对象,用自己的Observable包装并发出它.

比方说,我们有一个方法titleCase(String inputParam),它返回输入参数的Titled Cased String对象.此方法的返回类型可以是String或Observable<String>.

• 如果返回类型titleCase(..)仅仅是String,那么你会使用map(s -> titleCase(s))

• 如果返回类型titleCase(..)为是Observable<String>,那么你会使用flatMap(s -> titleCase(s))

希望澄清一下.

Answer 7

我只是想添加它flatMap,你真的不需要在函数内部使用自己的自定义Observable,你可以依赖标准的工厂方法/运算符:

Observable.from(jsonFile).flatMap(new Func1<File, Observable<String>>() {
    @Override public Observable<String> call(final File file) {
        try {
            String json = new Gson().toJson(new FileReader(file), Object.class);
            return Observable.just(json);
        } catch (FileNotFoundException ex) {
            return Observable.<String>error(ex);
        }
    }
});

通常,您应该尽可能避免从onXXX方法和回调中抛出(运行时)异常,即使我们在RxJava中放置了尽可能多的安全措施.

Answer 8

在该场景中使用map,您不需要新的Observable.

您应该使用Exceptions.propagate,它是一个包装器,因此您可以将这些已检查的异常发送到rx机制

Observable<String> obs = Observable.from(jsonFile).map(new Func1<File, String>() { 
    @Override public String call(File file) {
        try { 
            return new Gson().toJson(new FileReader(file), Object.class);
        } catch (FileNotFoundException e) {
            throw Exceptions.propagate(t); /will propagate it as error
        } 
    } 
});

然后,您应该在订户中处理此错误

obs.subscribe(new Subscriber<String>() {
    @Override 
    public void onNext(String s) { //valid result }

    @Override 
    public void onCompleted() { } 

    @Override 
    public void onError(Throwable e) { //e might be the FileNotFoundException you got }
};); 

有一个很好的帖子:http://blog.danlew.net/2015/12/08/error-handling-in-rxjava/