RxJava 操作符flatmap

有如下场景：

在前段调用后端的API时，经常会出现回调嵌套的情况。假设我们有两个API，queryA 和 queryB. 并且queryB的运行依赖于queryA的结果。那么我们的程序在一般的情况下可能是这个样子。

想象有如下的代码:





是不是感觉非常不舒服？假如嵌套的API再多几层，那么这将是个灾难。一个人开发的时候可能不觉得有什么问题，但是可以想象做code review或者新入项目组的同事看到你这复杂的嵌套时的表情。

是时候让flatmap出现啦！让我们把程序稍微改造一下，在Server类里面把makeRequest的方式变成RxJava的Observable的形式（我会在例子之后解释为什么要用flatmap()）:





看上去好像没觉得有都简洁是么？你等着我给你看看假如嵌套多几层之后:



看到了么？在RxJava的链式调用下，所有之前需要嵌套的地方都被flatMap()隔开了。代码可读性大大增加！假如你的IDE支持java 8的话，你可以体验更美妙的事情：lambda!



在抛弃了可恶的匿名类之后，代码更加简洁了！

看来很多同学都不太理解为何要用flatmap来解决嵌套回调。那咱们就深入点。

flatMap()究竟是用来干嘛的？简单的说，flatmap把一个Observable变成多个Observable，然后把得到的多个Obervable的元素一个个的发射出去。

假设你有一个API queryA，用这个query可以通过一个userID得到一个指定用户的所有关注的明星的userID(但是原始数据只是String)，你需要把这些userID一个个打印出来。

那么我们在调用queryA的时候就已经构建了一个Obervable了，我们暂且叫他O1.在O1每发射结果的同时，我们需要调用把返回的String结果变成另一个Observable，O2，O2含有所有的明星userID，并且一个个发射出去。代码例子：



大家可以看到，新的observable2是一个JsonObject类型的。因为在第三行注释之后，我们返回了一个(可以是多个)新的包含所有userID的observable，RxJava会将这个(或者多个)Observable平铺发射.

或者大家可以参考抛物线大神文章中的学生和课程的例子 大概在文章的中间

总之flatmap最核心的功能大家可以理解为：转换，可一个Obervable转换成多个Observable，再讲结果平铺发射。

在我们的例子中，是一个observable(O1)变成另一个observable(O2),是一对一的关系，因为queryA只会返回一个String s的结果，所以我们只会将一个observable（O2）平铺并发射。

所以这和nested callback有什么关系呢？

关系大了！

再想想，我们的刚刚做的和nested callback不同的地方差在哪里？唯一不通的地方是在获取所有明星userID的时候，我们是一个同步的操作，也就是说userID全部都包含在queryA返回的结果里面。我们用同步的方法把其封装成一个JSONArray.

假如queryA返回的只是用户关注明星userID的url呢？假如在第二部我们需要再做一步API call呢？

是的，就算我们还需要一步API call，程序的结构还是一样的：



flatmap最核心的功能大家可以理解为：转换，可一个Obervable转换成多个Observable，再将结果平铺发射。

至于转换的Observable里面的操作是同步还是异步，我们不需要去过多的思考，Rxjava已经完美的解决了。

参考：http://www.jianshu.com/p/0f926fda682b

延伸：实例项目（体会flatMap用法）

一. 添加依赖

compile 'com.squareup.retrofit2:retrofit:2.0.0-beta4'
compile 'io.reactivex:rxjava:1.1.1'

compile 'com.squareup.retrofit2:converter-gson:2.0.0-beta4'
//Gson解析器

compile 'io.reactivex:rxandroid:1.1.0'
//rxjava中用到的AndroidSchedulers.mainThread()

compile 'com.squareup.retrofit2:adapter-rxjava:2.0.0-beta4'
这里使用的是retrofit2.0,默认为okhttp3.0

二. 定义请求接口，转换HTTPAPI为Java接口

public interface IgankApi {
@GET("a.json")
Call<List<GirlEntity>> getGirl();

@GET("data/%E7%A6%8F%E5%88%A9/{count}/{page}")
Call<GirlJsonData> getGirl(@Path("count") int count, @Path("page") int page);

//与rxjava结合api
@GET("data/%E7%A6%8F%E5%88%A9/{count}/{page}")
Observable<GirlJsonData> getG(@Path("count") int count, @Path("page") int page);

}

三. 接着使用类Retrofit生成 接口的实现，使用了动态代理。

public static IgankApi getIgankApi() {

if (igankApi == null) {
synchronized (IgankApi.class) {
if (igankApi == null) {
Retrofit retrofit = new Retrofit.Builder().baseUrl("http://gank.io/api/")
.addConverterFactory(gsonConverterFactory)
.client(okHttpClient)
.addCallAdapterFactory(rxJavaCallAdapterFactory)
.build();
igankApi = retrofit.create(IgankApi.class);
}
}
}
return igankApi;
}

四. 调用接口

private void getImg() {
NetUtils.getIgankApi().getG(10, 2).flatMap(new Func1<GirlJsonData, Observable<List<GirlEntity>>>() {
@Override
public Observable<List<GirlEntity>> call(GirlJsonData girlJsonData) {
return Observable.just(girlJsonData.getResults());
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<List<GirlEntity>>() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(List<GirlEntity> girlEntities) {
girlEntityList.addAll(girlEntities);
loge(girlEntityList.get(1).getUrl() + "");
myRecycleViewAdapter.notifyDataSetChanged();
}
});

}

项目代码：https://github.com/jdsjlzx/Girl

再比如下面场景：

访问某个接口时并不能直接访问，而需要填入一个在线获取的 token ，代码应该怎么写？

Callback 方式，可以使用嵌套的 Callback：

@GET("/token")
public void getToken(Callback<String> callback);

@GET("/user")
public void getUser(@Query("token") String token, @Query("userId") String userId, Callback<User> callback);

...

getToken(new Callback<String>() {
@Override
public void success(String token) {
getUser(userId, new Callback<User>() {
@Override
public void success(User user) {
userView.setUser(user);
}

@Override
public void failure(RetrofitError error) {
// Error handling
...
}
};
}

@Override
public void failure(RetrofitError error) {
// Error handling
...
}
});

使用RxJava 的话，代码如下：

@GET("/token")
public Observable<String> getToken();

@GET("/user")
public Observable<User> getUser(@Query("token") String token, @Query("userId") String userId);

...

getToken()
.flatMap(new Func1<String, Observable<User>>() {
@Override
public Observable<User> onNext(String token) {
return getUser(token, userId);
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<User>() {
@Override
public void onNext(User user) {
userView.setUser(user);
}

@Override
public void onCompleted() {
}

@Override
public void onError(Throwable error) {
// Error handling
...
}
});

使用操作符flatMap() 就搞定了逻辑，依然是一条链。

flatMap()操作符的作用是将Observable发射的数据集合变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable，还是太抽象了。

简单说就是将 一个List 或者数组中的每一条数据都 转换成一个 Observable对象。

场景：

假如我们写了个网络请求， 然后拿出所有请求的数据， 现在我们不需要那么多数据， 我们只需要city字段和WD字段(运行过上段代码就知道了)， 而且这次我不在只给你一个url了，而是给你多个url。

那我们就开始写代码吧(代码是最好的老师)：

public static final String HOST = "http://www.weather.com.cn";
List<String> values = new ArrayList<>();
private String TAG = "SecondActivity2";

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
values.add("/adat/sk/101010100.html");
values.add("/adat/sk/101010100.html");
values.add("/adat/sk/101010100.html");
values.add("/adat/sk/101010100.html");
values.add("/adat/sk/101010100.html");

Observable.just(values).flatMap(new Func1<List<String>, Observable<?>>() {
@Override
public Observable<?> call(List<String> strings) {
return Observable.from(strings);
}
}).cast(String.class).map(new Func1<String, String>() {
@Override
public String call(String s) {
return doNetTaskForString(HOST + s); //取出想要的字段，这里我就不取出来了
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i(TAG, "value: " + s);
}
});
}

@NonNull
@Override
public int getContentView() {
return R.layout.activity_second2;
}

private synchronized String doNetTaskForString(String s) {

HttpClient client = new DefaultHttpClient();

Log.i(TAG, "url:" + s);
HttpGet get = new HttpGet(s);
String result;
try {
HttpResponse response = client.execute(get);
Log.i(TAG, "state code :" + response.getStatusLine().getStatusCode());
if (200 == response.getStatusLine().getStatusCode()) {
result = EntityUtils.toString(response.getEntity(), HTTP.UTF_8);
} else {
result = "状态行非200";
}

} catch (Exception e1) {
result = "抛出了异常" + e1.getMessage();
e1.printStackTrace();
}
return result;

}

打印log：



这段代码里我们又用到的新的操作符 cast .

cast的作用就是 在发射之前强制将Observable发射的所有数据转换为指定类型。



再举个例子。





这里结合了retrofit 在flatMap中通过一个城市的str,返回一个Observable,这个Observable的参数是一些获取到的天气信息结构WeatherData，这样在后面subscrib中就可以对其进行处理了。