Retrofit是一个很受欢迎的HTTP请求库，特别是在Android生态系统中。它既不是针对android的，也不是HTTP客户机的唯一选择。但是，因为它本身支持RxJava，所以对于移动应用程序来说，这是一个很好的选择，这两种方法都是用RxJava编写的，或者只愿意正确处理HTTP代码。在网络相关的工作代码中使用RxJava的主要优点是它能够轻松地在线程之间跳转。在我们开始进行改进之前，您需要以下依赖项。这个库本身就是RxJava的一个适配器，和一个Jackson JSON解析器:
compile 'com.squareup.retrofit2:retrofit:2.0.1'
compile 'com.squareup.retrofit2:adapter-rxjava:2.0.1'
compile 'com.squareup.retrofit2:converter-jackson:2.0.1'
通过要求您首先声明一个没有实现的Java接口，Retrofit改进了以安全的方式与RESTful服务交互。该接口随后被透明地转换为HTTP请求。为了练习的目的，我们将与Meetup API进行交互，它是一个组织事件的流行服务。其中一个端点返回给定位置附近的城市列表:
import retrofit2.http.GET;
import retrofit2.http.Query;
public interface MeetupApi {
@GET("/2/cities")
Observable<Cities> listCities(
@Query("lat") double lat,
@Query("lon") double lon
);
}
Retrofit 将会将对listCities() 方法的调用转化为一个网络调用。在引擎盖下，我们将会发出一个HTTP GET请求到/2/cities?lat=...&lon=...资源。注意返回类型。首先，我们有强类型的Cities，而不是字符串或弱类型的map-of-maps。但更重要的是，Cities来自一个Observable，当响应到达时，它会发出这个对象。Cities类基本上映射到我们从服务器接收的JSON中的大多数属性，settter/getter省略。
public class Cities {
private List<City> results;
}
public class City {
private String city;
private String country;
private Double distance;
private Integer id;
private Double lat;
private String localizedCountryName;
private Double lon;
private Integer memberCount;
private Integer ranking;
private String zip;
}
这种方法在抽象(使用高级概念，如方法调用和强类型响应)和底层细节(网络调用的异步本质)之间提供了良好的平衡。尽管HTTP具有请求-响应语义，但我们用Observable来建模不可避免的延迟，这样它就不会隐藏在一个泄漏的阻塞RPC(远程过程调用)的抽象的后面。不幸的是，为了与这个特定的API交互，您必须配置相当多的胶水代码。每个人状况可能不同，但重要的是要看到正确解析JSON响应所需的步骤：
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategy;
import retrofit2.Retrofit;
import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory;
import retrofit2.converter.jackson.JacksonConverterFactory;
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setPropertyNamingStrategy(
PropertyNamingStrategy.CAMEL_CASE_TO_LOWER_CASE_WITH_UNDERSCORES);
objectMapper.configure(
DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("https://api.meetup.com/")
.addCallAdapterFactory(
RxJavaCallAdapterFactory.create())
.addConverterFactory(
JacksonConverterFactory.create(objectMapper))
.build();
首先，我们需要对Jackson库中的ObjectMapper进行调优，以便将下划线字段名称无缝地转换为Java beans中使用的驼峰式大小写约定，例如，在City类中JSON中的localized_country_name以localizedCountryName转换。其次，我们希望避免在bean类中没有响应映射的字段。特别是JSON API往往是通过添加客户不支持的新字段来发展的。合理的默认是忽略这些字段，只使用对我们有意义的字段。因此，在不破坏现有客户机的情况下，服务器可以添加新的字段来响应。
有了一个改进的实例，我们最终可以综合使用MeetupApi实现来在整个客户端代码中使用:
MeetupApi meetup = retrofit.create(MeetupApi.class);
最后，通过我们的MeetupApi，我们可以制作一些HTTP请求并使用RxJava的功能。让我们构建一个更全面的示例。使用Meetup API，我们首先获取一个给定位置附近所有城市和城镇的列表:
double warsawLat = 52.229841;
double warsawLon = 21.011736;
Observable<Cities> cities = meetup.listCities(warsawLat, warsawLon);
Observable<City> cityObs = cities
.concatMapIterable(Cities::getResults);
Observable<String> map = cityObs
.filter(city -> city.distanceTo(warsawLat, warsawLon) < 50)
.map(City::getCity);
首先，我们使用concatMapIterable() .，将只有一个条目的Observable<Cities>
扩展为Observable<City>。然后，我们过滤出离初始位置50公里以内的城市。最后，我们提取城市名。我们的下一个目标是在Warsaw附近找到每个城市的人口，看看有多少人居住在半径50公里的范围内。要实现这一点，我们必须咨询GeoNames交付的另一个API。该方法通过给定的名称以及其他的属性，搜索某个位置，并返回人口数。哦们将在此使用Retrofit 来连接那个API：
public interface GeoNames {
@GET("/searchJSON")
Observable<SearchResult> search(
@Query("q") String query,
@Query("maxRows") int maxRows,
@Query("style") String style,
@Query("username") String username);
}
JSON对象必须映射到数据对象(getter和setter省略):
class SearchResult {
private List<Geoname> geonames = new ArrayList<>();
}
public class Geoname {
private String lat;
private String lng;
private Integer geonameId;
private Integer population;
private String countryCode;
private String name;
}
实例化GeoNames的方法类似于MeetupApi
GeoNames geoNames = new Retrofit.Builder()
.baseUrl("http://api.geonames.org")
.addCallAdapterFactory(RxJavaCallAdapterFactory.create())
.addConverterFactory(JacksonConverterFactory.create(objectMapper))
.build()
.create(GeoNames.class);
突然间，我们的示例应用程序使用了两种不同的api，并将它们均匀地合并在一起。对于每个城市名称，我们想咨询GeoNames API并提取人口:
Observable<Long> totalPopulation = meetup
.listCities(warsawLat, warsawLon)
.concatMapIterable(Cities::getResults)
.filter(city -> city.distanceTo(warsawLat, warsawLon) < 50)
.map(City::getCity)
.flatMap(geoNames::populationOf)
.reduce(0L, (x, y) -> x + y);
如果你仔细考虑一下，前面的程序在这个简洁的报表中做了大量的工作。首先，它要求MeetupApi提供一份城市名单，然后每一个城市都要获取人口。人口的响应(可能是异步的)在后面使用reduce()进行合计。最后，整个计算管道以Observable<Long>的形式结束，当所有城市的人口都积累起来时，就会产生一个long值。这显示了RxJava的真正力量，来自不同来源的流可以无缝地组合在一起。例如，populationOf()方法这是一个复杂的操作链,它将HTTP请求发送给GeoNames并以城市名抽取人口:
public interface GeoNames {
default Observable<Integer> populationOf(String query) {
return search(query)
.concatMapIterable(SearchResult::getGeonames)
.map(Geoname::getPopulation)
.filter(p -> p != null)
.singleOrDefault(0)
.doOnError(th ->
log.warn("Falling back to 0 for {}", query, th))
.onErrorReturn(th -> 0)
.subscribeOn(Schedulers.io());
}
default Observable<SearchResult> search(String query) {
return search(query, 1, "LONG", "some_user");
}
@GET("/searchJSON")
Observable<SearchResult> search(
@Query("q") String query,
@Query("maxRows") int maxRows,
@Query("style") String style,
@Query("username") String username
);
}
底层的通用的search()方法使用默认的方法进行包装，很容易使用。在收到用JSON包装的SearchResult对象之后，我们打解压出所有单独的搜索结果，确保在响应中没有出现人口，如果出现错误，我们只返回0。最后，我们确保在io() Scheduler上调用每个人口请求，以允许更好的并发性。subscribeOn()在这里实际上非常重要。没有它，每个城市的人口请求都将是串行的，大大增加了整个服务的延迟。然而，由于每个城市的flatMap()将调用populationOf()方法并在需要时订阅它，所以每个城市的数据都是同时获取的。实际上，我们还可以为每个用户请求添加一个timeout()操作符，从而以数据不完整的代价获得更好的响应时间。没有RxJava，实现此场景需要大量手工线程池集成，即使使用CompletableFuture ，这项任务也是很繁重的。然而，具有非入侵并发性和强大运算符的RxJava使编写快速和易于理解、简洁的代码成为可能。
结合两种不同的API，这两种api都是由Retrofit驱动的，多么富有魅力啊。然而，没有什么能阻止我们将完全不相关的Observable结合起来;例如，一个来自Retrofit, ，另一个来自JDBC调用，另一个来自接收JMS消息。所有这些用例都相当容易实现，既不泄漏抽象，也不会提供太多底层流实现的细节。