RxAndroidとは

RxAndroidはRxJavaのAndroid用の拡張モジュールです。
RxJavaは主に非同期処理を行うことが出来るライブラリです。
またObserverパターンに踏襲しているため、サードパーティーのライブラリとの連携も可能です。
ECMA6からJavascriptに標準搭載される予定のPromiseみたいなものです。
ここではPromise.whenみたいなことをやろうとしてます。
jQueryを使うとこんな感じです。javascriptは簡単ですね

var baseUrl = "http://api.openweathermap.org/data/2.5/weather?q=";
$.when(
    $.getJSON(baseUrl+"Tokyo"),
    $.getJSON(baseUrl+"Kyoto"),
    $.getJSON(baseUrl+"Osaka"),
    $.getJSON(baseUrl+"Okinawa"),
    $.getJSON(baseUrl+"Hokkaido")
)
.then(
    function(data1, data2, data3, data4, data5){
        var s = "";
        s += data1[0].id + ":" + data1[0].name + "\n";
        s += data2[0].id + ":" + data2[0].name + "\n";
        s += data3[0].id + ":" + data3[0].name + "\n";
        s += data4[0].id + ":" + data4[0].name + "\n";
        s += data5[0].id + ":" + data5[0].name + "\n";
        return $.Deferred().resolve(s);
    }
)
.then(
    function(s){
        console.log(s);
    }
);

ちょっと思想が違いますが、競合はBoltsFrameworkとかですかね。
BoltsFrameworkは生理的に受け付けないので、こちらを推していきたいところです。
ここではRxAndroidとRESTAPIライブラリのRetrofitを利用して、
APIへ並列でアクセスするプログラムを書いてみます。
APIはopenweathermapを利用します。

build.gradle

    compile 'com.squareup.retrofit:retrofit:1.9.0'
    compile 'io.reactivex:rxandroid:1.0.1'

API用のinterfaceを作成

Retrofitを利用します。アノテーションで色々指定出来て便利です。
結果はデータクラスを作るのが面倒なのでJSONObjectを利用します。

public interface WeatherApi {
    @GET("/data/2.5/weather")
    public Observable<JSONObject> get(@Query("q") String query);
}

JsonConverterの作成

標準だとGsonConverterしかないので、作成します。
- GsonConverを利用する場合は必要ないです。

public class JsonConverter implements Converter {
    private static final Charset UTF_8 = Charset.forName("UTF-8");

    @Override
    public Object fromBody(TypedInput body, Type type) 
        throws ConversionException {
        InputStream inputStream = null;
        JSONObject jsonObject = null;
        try {
            inputStream = body.in();
            jsonObject = new JSONObject(IOUtils.toString(inputStream, UTF_8));
        } catch (IOException | JSONException e) {
            e.printStackTrace();
        } finally {
            try {
                if (inputStream != null) {
                    inputStream.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return jsonObject;
    }

    @Override
    public TypedOutput toBody(Object object) {
        return null;
    }
}

APIを作成する

API生成時にJsonConverterを渡しています。
LogLevel.FULLだとすべての通信ログが確認できます。

WeatherApi api = new RestAdapter.Builder()
    .setConverter(new JsonConverter())
    .setEndpoint("http://api.openweathermap.org/")
    .setLogLevel(RestAdapter.LogLevel.FULL)
    .build()
    .create(WeatherApi.class);

並列でAPIを呼んでみる

Observable.zipを利用します。
環境の都合上、ラムダ式を利用しています。
TextViewに表示するための文字列を作って返しています。

Observable.zip(
    api.get("Tokyo"), 
    api.get("Kyoto"), 
    api.get("Osaka"), 
    api.get("Okinawa"), 
    api.get("Hokkaido"),
    (jsonObject, jsonObject2, jsonObject3, jsonObject4, jsonObject5) -> {
        String newLine = System.getProperty("line.separator");
        StringBuilder sb = new StringBuilder();
        try {
            sb.append(jsonObject.getString("id")+":"+jsonObject.getString("name"));
            sb.append(newLine);
            sb.append(jsonObject2.getString("id")+":"+jsonObject2.getString("name"));
            sb.append(newLine);
            sb.append(jsonObject3.getString("id")+":"+jsonObject3.getString("name"));
            sb.append(newLine);
            sb.append(jsonObject4.getString("id")+":"+jsonObject2.getString("name"));
            sb.append(newLine);
            sb.append(jsonObject5.getString("id")+":"+jsonObject3.getString("name"));
            sb.append(newLine);
        } catch (JSONException e) {
            e.printStackTrace();
        }
        return sb.toString();
    })
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Subscriber<String>() {
        @Override
        public void onCompleted() {
        }
        @Override
        public void onError(Throwable e) {
            e.printStackTrace();
            Toast.makeText(this, "error", Toast.LENGTH_SHORT).show();
        }
        @Override
        public void onNext(String s) {
            result.setText(s); // 表示するTextView
        }
    });

subscribeOn と observerOn

subscribeOn(Schedulers.newThread())は
Observable.zipの処理を新規スレッドで実行することを表します。
AndroidSchedulers.mainThread()を渡すことも可能。

observeOn(AndroidSchedulers.mainThread())は
subscriberの処理をMainスレッドで実行することを表します。

parallel-rxandroid-example

REST APIにRxAndroid とretrofitで並列にアクセスしてみる