RxJava(十三)：RxAndroid 2.x 和 Retrofit 的使用

時間 2020-06-29

標籤 rxjava 十三 rxandroid 2.x retrofit 使用欄目 Java 简体版

原文原文鏈接

1. RxAndroid 2.x 簡介

1.1 介紹

近幾年 RxJava 逐漸成爲 Android 開發的新寵，愈來愈多的 Android 開發者正在使用或者即將使用 RxJava 。要想在 Android 上使用 RxJava, RxAndroid 必不可少.react

RxAndroid GitHub 地址android

https://github.com/ReactiveX/RxAndroid

RxAndroid 也隸屬於 ReactiveX 組織，它是 RxJava 在 Android 上的一個擴展。從其 GitHub 的官方主頁上，能夠了解到 RxAndroid 提供了一個調度程序，可以切換到 Android 主線程或者任意指定的 Looper。git

RxAndroid 不能單獨使用，它只有依賴 RxJava 才能使用。注意，選擇所依賴的 RxJava 版本時最好用最新的版本，由於新版本會修復以前版本的 Bug 而且提供一些新的特性。github

RxAndroid 的下載：編程

implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
// Because RxAndroid releases are few and far between, it is recommended you also
// explicitly depend on RxJava's latest version for bug fixes and new features.
// (see https://github.com/ReactiveX/RxJava/releases for latest 2.x.x version)
implementation 'io.reactivex.rxjava2:rxjava:2.x.x'

1.2 使用

1.2.1 AndroidSchedulers.mainThread()

在 Android 中使用時須要新增一個調度器，用於將指定的操做切換到 Android 的主線程中運行，方便作一些更新 UI 的操做。 RxAndroid 就提供了這樣的調度器。json

下面例子展現了從網上獲取圖片而後將圖片轉換成 Bitmap，最後顯示到 ImageView 上的
過程。這裏沒有使用 Android 的 AsyncTask 或 Handler，而是經過 subscribeOn, observeOn 不斷地切換線程來達到目的的。segmentfault

Observable.create(new ObservableOnSubscribe<Bitmap>() {
    @Override
    public void subscribe(ObservableEmitter<Bitmap> emitter) throws Exception {
        Log.d(TAG, "加載Bitmap的線程名：" + Thread.currentThread().getName());
        emitter.onNext(getBitmap());
    }
})      // 設置數據加載在子線程中進行
        .subscribeOn(Schedulers.io())
        // 設置圖片在主線程中進行
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<Bitmap>() {
            @Override
            public void accept(Bitmap bitmap) throws Exception {
                Log.d(TAG, "Next-> 更新UI的線程名: " + Thread.currentThread().getName());
                if (bitmap != null) {
                    imageView.setImageBitmap(bitmap);
                } else {
                    Log.d(TAG, "Next-> 加載圖片失敗. 線程名：" + Thread.currentThread().getName());
                }
            }
        });

private Bitmap getBitmap() {
    HttpURLConnection connection = null;
    try {
        URL url = new URL("http://www.designerspics.com/wp-content/uploads/2014/09/grass_shrubs_2_free_photo.jpg");

        connection = (HttpURLConnection) url.openConnection();
        connection.setConnectTimeout(20000);
        connection.connect();

        if (connection.getResponseCode() == 200) {
            return BitmapFactory.decodeStream(connection.getInputStream());
        }
    } catch (Exception e) {
        Log.d(TAG, "getBitmap: " + e.getMessage());
    } finally {
        if (connection != null) connection.disconnect();
    }
    return null;
}

// Log日誌輸出結果
 加載Bitmap的線程名：RxCachedThreadScheduler-1
 Next-> 更新UI的線程名: main

1.2.2 AndroidSchedulers.from(Looper looper)

RxAndroid 除了能夠切換到主線程，還可使用任意指定的 Looperapi

Observable 建立以後先發射圖片的 URL，而後再在 map 操做中獲取圖片的 Bitmap，最後將 Bitmap 顯示到 ImageView緩存

final String image_url = "http://www.designerspics.com/wp-content/uploads/2014/09/grass_shrubs_2_free_photo.jpg";

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        Log.d(TAG, "發射的線程名:" + Thread.currentThread().getName());
        emitter.onNext(image_url);
    }
}).subscribeOn(AndroidSchedulers.from(Looper.getMainLooper()))
        .observeOn(Schedulers.io())
        .map(new Function<String, Bitmap>() {
            @Override
            public Bitmap apply(String s) throws Exception {
                Log.d(TAG, "轉換成Bitmap的線程名:" + Thread.currentThread().getName());
                return getBitmap(s);
            }
        })
         // 設置圖片在主線程中進行
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<Bitmap>() {
            @Override
            public void accept(Bitmap bitmap) throws Exception {
                Log.d(TAG, "Next-> 更新UI的線程名: " + Thread.currentThread().getName());
                if (bitmap != null) {
                    imageView.setImageBitmap(bitmap);
                } else {
                    Log.d(TAG, "Next-> 加載圖片失敗. 線程名：" + Thread.currentThread().getName());
                }
            }
        });

// Log日誌輸出結果
 發射的線程名:main
 轉換成Bitmap的線程名:RxCachedThreadScheduler-1
 Next-> 更新UI的線程名: main

2. Retrofit 簡介

Retrofit 是一個在 Android 開發中很是流行的網絡框架，底層依賴 OkHttp。 Retrofit 和 OkHttp
都出自 Square 的技術團隊。

Retrofit 的 GitHub 地址

https://github.com/square/retrofit

應用程序經過 Retrofit 請求網絡，其實是使用 Retrofit 接口層封裝請求參數、 Header、URL 等信息，以後由 OkHttp 完成後續的請求操做，在服務端返回數據以後， OkHttp 將原始的結果交給 Retrofit, Retrofit 再根據用戶的需求對結果進行解析的過程。

Retrofit 支持大多數的 Http 方法。

Retrofit 的特色以下：
(1) Retrofit 是可插拔的，容許不一樣的執行機制及其庫用於執行 http 調用。容許 API 請求，與應用程序其他部分中任何現有線程模型或任務框架無縫組合。

Retrofit 爲常見的框架提供了適配器 ( Adapter ):

RxJaval.x Observable & Single - com.squareup.retrofit2:adapter-rxjava
RxJava2.x Observable, Flowable, Single, Completable & Maybe - com.squareup.retrofit2:adapter-rxjava2
Guava ListenableFuture - com.squareup.retrofit2:adapter-guava
Java 8 CompletableFuture - com.squareup.retrofit2:adapter-java8

(2) ）容許不一樣的序列化格式及其庫，用於將 Java 類型轉換爲其 http 表示形式，並將 http 實體解析爲 Java 類型。

Retrofit 爲常見的序列化格式提供了轉換器 ( Converter ):

Gson: com.squareup.retrofit2:converter-gson
Jackson: com.squareup.retrofit2:converter-jackson
Moshi: com.squareup.retrofit2:converter-moshi
Protobuf: com.squareup.retrofit2:converter-protobuf
Wire: com.squareup.retrofit2:converter-wire
Simple XML: com.squareup.retrofit2:converter-simplexml
Scalars (primitives, boxed, and String): com.squareup.retrofit2:converter-scalars

開源社區也己經爲其餘庫和序列化格式建立了各類第三方轉換器 ( Converter ):

LoganSquare - com.github.aurae.retrofit2:converter-logansquare:1.4.1
FastJson - org.ligboy.retrofit2:converter-fastjson:2.1.0 和 org.ligboy.retrofit2:converter-fastjson-android:2.1.0

OkHttp 的特色以下：

支持 HTTP2/SPDY 黑科技
socket 自動選擇最優路線，並支持自動重連。
擁有自動維護的 socket 鏈接池，減小握手次數
擁有隊列線程池，輕鬆寫併發
擁有 Interceptors 輕鬆處理請求與響應（好比透明 GZIP 壓縮、 LOGGING）
基於 Headers 的緩存策略。

3. Retrofit 與 RxJava 的完美配合

Retrofit 是一個網絡框架，若是想嘗試響應式的編程方式，則能夠結合 RxJava 一塊兒使用。Retrofit 對 RxJava1 和 RxJava2 都提供了 Adapter。

案例：將蘇州市南門地區的 PM2.五、PM十、SO2 的數據展現到 App 上。在 http://pm25.in/上能夠找到獲取...， pm25.in 是一個公益性的網站，免費提供空氣質量數據。在調用這些接口以前，
須要去該網站註冊，並申請一個 AppKey

Retrofit 使用步驟以下：

添加 Retrofit 依賴

在 App 的 build.gradle 中添加所須要的 Retrofit 庫，以及 RxJava2 的 adapter 庫。

implementation 'com.squareup.retrofit2:retrofit:2.7.1'
implementation 'com.squareup.retrofit2:adapter-rxjava2:2.7.1'

implementation 'org.ligboy.retrofit2:converter-fastjson-android:2.1.0'

implementation "com.squareup.okhttp3:logging-interceptor:4.3.1"

建立 RetrofitManager

通常須要建立 Retrofit 管理類，在這裏建立一個名爲 RetrofitManager 類，方便在
整個 App 中使用。

RetrofitManager 代碼以下：

public class RetrofitManager {

    private static Retrofit retrofit;

    public static Retrofit retrofit() {
        if (retrofit == null) {
            HttpLoggingInterceptor loggingInterceptor = new HttpLoggingInterceptor()
                    .setLevel(HttpLoggingInterceptor.Level.BASIC);

            OkHttpClient okHttpClient = new OkHttpClient.Builder()
                    .writeTimeout(30_1000, TimeUnit.MILLISECONDS)
                    .readTimeout(20_1000, TimeUnit.MILLISECONDS)
                    .connectTimeout(15_1000, TimeUnit.MILLISECONDS)
                    .addInterceptor(loggingInterceptor)
                    .build();

            retrofit = new Retrofit.Builder()
                    .baseUrl(APIService.API_BASE_SERVER_URL)
                    .addConverterFactory(FastJsonConverterFactory.create())
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .client(okHttpClient)
                    .build();
        }

        return retrofit;
    }
}

建立 APIService

接下來，咱們須要定義網絡請求的接口。 pm25.in 提供了多個獲取空氣質量數據的接口，這裏選取其中 3 個接口，分別是獲取一個城市全部監測點的PM2.5數據、獲取一個城市全部監測點的PM10數據、獲取一個城市全部監測點的 SO2 數據接口。

public interface APIService {
    String API_BASE_SERVER_URL = "http://www.pm25.in/";

    @GET("api/querys/pm2_5.json")
    Maybe<List<PM25Model>> pm25(@Query("city") String city, @Query("token") String token);

    @GET("api/querys/pm10.json")
    Maybe<List<PM10Model>> pm10(@Query("city") String city, @Query("token") String token);

    @GET("api/querys/so2.json")
    Maybe<List<SO2Model>> so2(@Query("city") String city, @Query("token") String token);
}

在 APIService 中，每一個方法返回的類型都是 Maybe 類型，其實也能夠返回 Observable、Flowable 等類型。

Retrofit 的使用

下面的代碼分別調用了 3 個接口，井過濾出了南門地區的相關數據。

APIService apiService = RetrofitManager.retrofit().create(APIService.class);

apiService.pm25(Constant.CITY, Constant.TOKEN)
        .compose(RxJavaUtils.<List<PM25Model>>maybeToMain())
        .filter(new Predicate<List<PM25Model>>() {
            @Override
            public boolean test(List<PM25Model> pm25Models) throws Exception {
                return pm25Models != null && !pm25Models.isEmpty();
            }
        })
        .flatMap(new Function<List<PM25Model>, MaybeSource<PM25Model>>() {
            @Override
            public MaybeSource<PM25Model> apply(List<PM25Model> pm25Models) throws Exception {
                for (PM25Model pm25Model : pm25Models) {
                    if ("南門".equals(pm25Model.position_name)) {
                        return Maybe.just(pm25Model);
                    }
                }
                return null;
            }
        })
        .subscribe(new Consumer<PM25Model>() {
            @Override
            public void accept(PM25Model pm25Model) throws Exception {
                Log.d(TAG, "PM25.Success-> " + pm25Model);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(TAG, "PM25.Error-> " + throwable);
            }
        });

apiService.pm10(Constant.CITY, Constant.TOKEN)
        .compose(RxJavaUtils.<List<PM10Model>>maybeToMain())
        .filter(new Predicate<List<PM10Model>>() {
            @Override
            public boolean test(List<PM10Model> pm10Models) throws Exception {
                return pm10Models != null && !pm10Models.isEmpty();
            }
        })
        .flatMap(new Function<List<PM10Model>, MaybeSource<PM10Model>>() {
            @Override
            public MaybeSource<PM10Model> apply(List<PM10Model> pm10Models) throws Exception {
                for (PM10Model pm10Model : pm10Models) {
                    if ("南門".equals(pm10Model.position_name)) {
                        return Maybe.just(pm10Model);
                    }
                }
                return null;
            }
        })
        .subscribe(new Consumer<PM10Model>() {
            @Override
            public void accept(PM10Model pm10Model) throws Exception {
                Log.d(TAG, "PM10.Success-> " + pm10Model);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(TAG, "PM10.Error-> " + throwable);
            }
        });

apiService.so2(Constant.CITY, Constant.TOKEN)
        .compose(RxJavaUtils.<List<SO2Model>>maybeToMain())
        .filter(new Predicate<List<SO2Model>>() {
            @Override
            public boolean test(List<SO2Model> so2Models) throws Exception {
                return so2Models != null && !so2Models.isEmpty();
            }
        })
        .flatMap(new Function<List<SO2Model>, MaybeSource<SO2Model>>() {
            @Override
            public MaybeSource<SO2Model> apply(List<SO2Model> so2Models) throws Exception {
                for (SO2Model so2Model : so2Models) {
                    if ("南門".equals(so2Model.position_name)) {
                        return Maybe.just(so2Model);
                    }
                }
                return null;
            }
        })
        .subscribe(new Consumer<SO2Model>() {
            @Override
            public void accept(SO2Model so2Model) throws Exception {
                Log.d(TAG, "SO2.Error-> " + so2Model);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(TAG, "SO2.Error-> " + throwable);
            }
        });

這裏還使用了 maybeToMain() 方法，它的代碼以下：

@JvmStatic
fun <T> maybeToMain(): MaybeTransformer<T, T> {
    return MaybeTransformer { upstream ->
        upstream.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
    }
}

它用於切換線程，返回 MaybeTransformer 對象。由於 apiService 中每一個返回的方法都是
Maybe 類型，因此這裏會用到 MaybeTransformer 。使用了 maybeToMain() 後，除網絡請求是在
io() 線程中運行外，其他的操做都是在主線程中運行的.

也可讓 filter、 flatMap 操做也在 io() 線程中運行，展現數據時才切換回主線程。

接下來列舉一些 Retrofit 其他常見的使用場景。

(1) 合併多個網絡請求

如：須要在某一個信息流列表中插入多條廣告，每一條廣告都須要作一次網絡請求。這時就能夠考慮使用 zip 操做符，將請求信息流，以及請求的多個廣告的請求合併起來，等全部請求完成以後，再用合併函數將廣告插到信息流固定的位置上，最後以列表的形式呈現給用戶。

APIService apiService = RetrofitManager.retrofit().create(APIService.class);

Maybe<PM25Model> pm25ModelMaybe = apiService.pm25(Constant.CITY, Constant.TOKEN)
        .compose(RxJavaUtils.<List<PM25Model>>maybeToMain())
        .filter(new Predicate<List<PM25Model>>() {
            @Override
            public boolean test(List<PM25Model> pm25Models) throws Exception {
                return pm25Models != null && !pm25Models.isEmpty();
            }
        })
        .flatMap(new Function<List<PM25Model>, MaybeSource<PM25Model>>() {
            @Override
            public MaybeSource<PM25Model> apply(List<PM25Model> pm25Models) throws Exception {
                for (PM25Model pm25Model : pm25Models) {
                    if ("南門".equals(pm25Model.position_name)) {
                        return Maybe.just(pm25Model);
                    }
                }
                return null;
            }
        });

Maybe<PM10Model> pm10ModelMaybe = apiService.pm10(Constant.CITY, Constant.TOKEN)
        .compose(RxJavaUtils.<List<PM10Model>>maybeToMain())
        .filter(new Predicate<List<PM10Model>>() {
            @Override
            public boolean test(List<PM10Model> pm10Models) throws Exception {
                return pm10Models != null && !pm10Models.isEmpty();
            }
        })
        .flatMap(new Function<List<PM10Model>, MaybeSource<PM10Model>>() {
            @Override
            public MaybeSource<PM10Model> apply(List<PM10Model> pm10Models) throws Exception {
                for (PM10Model pm10Model : pm10Models) {
                    if ("南門".equals(pm10Model.position_name)) {
                        return Maybe.just(pm10Model);
                    }
                }
                return null;
            }
        });

Maybe<SO2Model> so2ModelMaybe = apiService.so2(Constant.CITY, Constant.TOKEN)
        .compose(RxJavaUtils.<List<SO2Model>>maybeToMain())
        .filter(new Predicate<List<SO2Model>>() {
            @Override
            public boolean test(List<SO2Model> so2Models) throws Exception {
                return so2Models != null && !so2Models.isEmpty();
            }
        })
        .flatMap(new Function<List<SO2Model>, MaybeSource<SO2Model>>() {
            @Override
            public MaybeSource<SO2Model> apply(List<SO2Model> so2Models) throws Exception {
                for (SO2Model so2Model : so2Models) {
                    if ("南門".equals(so2Model.position_name)) {
                        return Maybe.just(so2Model);
                    }
                }
                return null;
            }
        });


Maybe.zip(pm25ModelMaybe, pm10ModelMaybe, so2ModelMaybe, new Function3<PM25Model, PM10Model, SO2Model, ZipObject>() {
    @Override
    public ZipObject apply(PM25Model pm25Model, PM10Model pm10Model, SO2Model so2Model) throws Exception {
        Log.d(TAG, "zip-> \r\n" + pm25Model + "\r\n" + pm10Model + "\r\n" + so2Model);
        ZipObject zipObject = new ZipObject();

        zipObject.pm2_5 = pm25Model.pm2_5;
        zipObject.pm2_5_24h = pm25Model.pm2_5_24h;
        zipObject.pm2_5_quality = pm25Model.quality;

        zipObject.pm10 = pm10Model.pm10;
        zipObject.pm10_24h = pm10Model.pm10_24h;

        zipObject.so2 = so2Model.so2;
        zipObject.so2_24h = so2Model.so2_24h;

        return zipObject;
    }
}).subscribe(new Consumer<ZipObject>() {
    @Override
    public void accept(ZipObject zipObject) throws Exception {
        Log.d(TAG, "Success-> " + zipObject);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable throwable) throws Exception {
        Log.d(TAG, "Error-> " + throwable);
    }
});

(2) 返回默認值

有時, 網絡請求失敗可使用 onErrorReturn 操做符, 一個空的對象做爲默認值。

(3) 多個網絡請求嵌套使用

如果 A 請求完成以後，才能去調用 B 請求，則能夠考慮使用 flatMap 操做符。

(4) 重試機制

對於一些重要的接口，須要採用重試機制。由於有些時候用戶的網絡環境比較差，第一次請求接口超時了，那麼再一次請求可能就會成功。雖然有必定的延時，但至少返回了數據，保證了用戶體驗。

apiService.loadContent(params)
        .retryWhen(new RetryWithDelay(3, 1000));

在這裏 retryWhen 操做符與 RetryWithDelay 一塊兒搭配使用，表示有 3 次重試機會，每次的延遲時間是 1000ms。 RetryWithDelay 是一個工具類，使用kotlin語言編寫。

class RetryWithDelay(
    private val maxRetries: Int,
    private val retryDelayMillis: Int
) : Function<Flowable<out Throwable>, Publisher<*>> {

    private var retryCount: Int = 0

    init {
        this.retryCount = 0
    }

    override fun apply(attempts: Flowable<out Throwable>): Publisher<*> {
        return attempts.flatMap { throwable ->
            if (++retryCount <= maxRetries) {
                Flowable.timer(retryDelayMillis.toLong(), TimeUnit.MILLISECONDS)
            } else {
                Flowable.error(throwable)
            }
        }
    }
}

若是個人文章對您有幫助，不妨點個贊鼓勵一下(^_^)