使用Retrofit2+RxJava2+ProtoBuf實現網絡請求

時間 2019-11-08

標籤使用 retrofit2+rxjava2+protobuf retrofit rxjava protobuf 實現網絡請求欄目 Java 简体版

原文原文鏈接

引言

Retrofit 是一個用於 Android 和 Java 平臺的類型安全的，底層使用OkHttp實現網絡請求框架。Retrofit 經過將 API 抽象成 Java 接口而讓咱們鏈接到 REST web 服務變得很輕鬆。
RxJava 提供一套異步編程的 API，這套 API 是基於觀察者模式的，並且是鏈式調用的。
Protocol Buffers 是一種輕便高效的結構化數據存儲格式，能夠用於結構化數據串行化，或者說序列化。它很適合作數據存儲或 RPC 數據交換格式。
主要講解如何使用各個庫封裝網絡請求，不講解各庫如何使用，具體可查看Rxjava2、Retrofit2 、ProtoBuf。java

依賴

implementation 'com.squareup.retrofit2:retrofit:2.5.0'
implementation 'io.reactivex.rxjava2:rxjava:2.2.4'
implementation 'com.squareup.wire:wire-runtime:2.3.0-RC1'
implementation 'com.squareup.retrofit2:retrofit-adapters:2.5.0'
implementation 'com.squareup.retrofit2:converter-wire:2.5.0'
implementation 'com.squareup.retrofit2:adapter-rxjava2:2.5.0'
implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'

其中RxAndroid是爲優雅地處理異步請求和解決線程調度問題；Wire用於將請求結果轉換爲實體類型，而且wire是生成ProtoBuf文件的一種，沒有用官方的protobuf生成java文件，主要是爲了解決64k限制，減小生成java的代碼量，但須要注意的是wire生成的java文件須要判斷null，而官方的protobuf生成java文件是有默認值的，無需判斷null。react

proto文件

在客戶端可能須要相關客戶端信息，例如設備信息、設備類型，app信息、網絡信息等。android

// 設備類型
enum PBDeviceType {
    DEVICE_ANDROID = 0;                      // 安卓
    DEVICE_IOS = 1;                          // 蘋果
    DEVICE_PC = 2;                           // PC
}

// 設備
message PBDevice {
    string deviceId = 1;                    // 設備ID
    string deviceOs = 2;                    // 設備操做系統
    string deviceModel = 3;                 // 設備模型
    PBDeviceType deviceType = 4;             // 設備類型，參考PBDeviceType
}

// 網絡類型
enum PBNetworkType {
    NET_UNKNOWN = 0;                         // 未知網絡
    NET_WIFI = 1;                            // WIFI
    NET_2G = 2;                              // 2G網絡
    NET_3G = 3;                              // 3G網絡
    NET_4G = 4;                              // 4G網絡
}

// APP信息
message PBAppInfo {
    string versionName = 1;                 // 應用程序版本名
    uint32 versionCode = 2;                 // 應用程序版本號
    PBNetworkType network = 3;               // 網絡信息
    PBDevice device = 4;                     // 設備信息
}

定義Request和Responsegit

// 消息請求包
message PBRequest {
    uint32 type = 1;                        // 消息類型
    bytes messageData = 2;                  // 請求數據
    uint64 timestamp = 3;                   // 客戶端時間戳
    PBAppInfo appInfo = 4;                   // APP信息
}

// 消息響應包
message PBResponse {
    uint32 type = 1;                        // 消息類型
    bytes messageData = 2;                  // 返回數據
    uint32 resultCode = 3;                  // 返回的結果碼
    string resultInfo = 4;                  // 返回的結果消息提示文本（用於錯誤提示）
}

使用命令生成相關的Java文件github

java -jar ${wire_compiler} --proto_path=${protoPath} --java_out=${modelPath} $1

wire_compiler：wire.jar的存放路徑
protoPath：proto文件存放路徑
modelPath：生成java文件存放路徑

請求接口

定義一個Service的接口類，管理全部請求接口web

public interface Service {

    @POST("users/new")
    Observable<Response<PBResponse>> sendMessage(@Body PBRequest request);
}

請求處理

定義一個RetrofitHandler類，處理retrofit的初始化和發送請求，該類使用靜態單例進行初始化。編程

private Service mService;           // 請求接口
private static String mServiceUrl;  // 請求url

private static class Holder {
    private static final RetrofitHandler INSTANCE = new RetrofitHandler();
}

private RetrofitHandler() {
    init();
}

public static RetrofitHandler getInstance(String serviceUrl) {
    mServiceUrl = serviceUrl;
    return Holder.INSTANCE;
}

retrofit的初始化，使用wire的factory進行數據轉換，使用rxjava2進行適配。api

private void init() {
    assert mServiceUrl != null;
    Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(mServiceUrl)
            .client(createClient())
            .addConverterFactory(WireConverterFactory.create())
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .build();
    mService = retrofit.create(Service.class);
}

private OkHttpClient createClient() {
    return new OkHttpClient.Builder()
            .retryOnConnectionFailure(true)
            .connectionPool(new ConnectionPool(5, 1, TimeUnit.MINUTES))
            .connectTimeout(10, TimeUnit.SECONDS)
            .readTimeout(20, TimeUnit.SECONDS)
            .writeTimeout(5, TimeUnit.SECONDS)
            .build();

}

retrofit進行接口調用發送請求安全

public Observable<PBResponse> send(final PBRequest request) {
    return mService.sendMessage(request)
            .map(new Function<Response<PBResponse>, PBResponse>() {
                @Override
                public PBResponse apply(Response<PBResponse> response) throws Exception {
                    if (response == null) {
                        return failed(request.type, 1001, "未知錯誤");
                    }
                    if (response.code() != 200 || response.body() == null) {
                        return failed(request.type, response.code(), response.message());
                    }
                    return response.body();
                }
            });
}

private PBResponse failed(int type, int code, String info) {
    return new PBResponse.Builder()
            .type(type)
            .resultCode(code)
            .resultInfo(info)
            .build();
}

返回處理

定義ApiResult的泛型返回模型，對返回數據進行二次處理網絡

public class ApiResult<T extends Message> {
    private int code;        // 返回碼
    private String message;  // 返回信息
    private T response;      // 返回數據

    public ApiResult(int code, String message, T response) {
        this.code = code;
        this.message = message;
        this.response = response;
    }

    public int getCode() {
        return code;
    }

    public String getMessage() {
        return message;
    }

    public T getResponse() {
        return response;
    }
}

定義ApiCallback泛型回調處理類，給業務層進行相關的業務處理，如刷新UI等。

public class ApiCallback<T extends Message> {
    private boolean isDisposed = false;  // 返回處理標誌

    public boolean isDisposed() {
        return isDisposed;
    }

    public void setDisposed(boolean disposed) {
        isDisposed = disposed;
    }

    public void onStart() {
       // 請求開始
    }

    public void onSuccess(T response) {

    }

    public void onFailure(int code, String message) {

    }

    public void onCompleted() {
        // 請求完成
    }
}

定義IRquest接口，給業務端初始化相關客戶端信息

public interface IRequest {
    String getVersionName();

    Integer getVersionCode();

    PBNetworkType getNetworkType();

    PBDevice getDevice();

}

定義HttpManager類，使用單例進行實例化

private static class Holder {
    private static final HttpManager INSTANCE = new HttpManager();
}

private HttpManager() {

}

public static HttpManager getInstance() {
    return HttpManager.Holder.INSTANCE;
}

定義一個初始化方法，能夠在application進行初始化

public void init(IRequest request, String url){
    mRequest = request;
    mHandler = RetrofitHandler.getInstance(url);
}

使用Rxjava2進行返回處理

@SuppressLint("CheckResult")
public <T extends Message, P extends Message> Disposable request(final T request, final int messageType,
                                                                 final Class<P> pClass, final ApiCallback<P> apiCallback) {
    
    Observable observable = mHandler.send(buildBody(request, messageType))  // 發送請求
            .map(new Function<PBResponse, ApiResult<? extends Message>>() {
                @Override
                public ApiResult<? extends Message> apply(PBResponse pbResponse) throws Exception {
                    return apiResult(pClass, pbResponse);
                }
            });
    observable.doOnDispose(new Action() { // 業務端取消訂閱時調用
        @Override
        public void run() throws Exception {
            apiCallback.setDisposed(true);
        }
    });
    return toSubscribe(observable, new Consumer<ApiResult<P>>() {
        @Override
        public void accept(ApiResult<P> apiResult) throws Exception {
            if (apiCallback == null || apiCallback.isDisposed()) return;
            try {
                if (apiResult.getCode() == 200) {
                    apiCallback.onSuccess(apiResult.getResponse());
                } else {
                    apiCallback.onFailure(apiResult.getCode(), apiResult.getMessage());
                }
            } catch (Exception ex) {
                apiCallback.onFailure(1004, "客戶端處理異常");
            } finally {
                apiCallback.onCompleted();
            }
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) {
            if (apiCallback == null || apiCallback.isDisposed()) return;
            //客戶端本地的異常，如斷網等
            try {
                apiCallback.onFailure(1005, "網絡鏈接錯誤");
            } catch (Exception e) {
            } finally {
                apiCallback.onCompleted();
            }
        }
    }, new Action() {
        @Override
        public void run() {
            if (apiCallback == null || apiCallback.isDisposed()) return;

            //onCompleted will not be called when occurs network exception, like disconnected/timeout, replace invoking at onNext/onError
        }
    }, new Consumer<Disposable>() {
        @Override
        public void accept(Disposable disposable) {
            if (apiCallback == null || apiCallback.isDisposed()) return;
            apiCallback.onStart();
        }
    });
}

// 異步訂閱
private <T> Disposable toSubscribe(Observable<T> o, Consumer<T> onNext, Consumer<Throwable> onError, Action onComplete, Consumer<Disposable> onSubscribe) {
    return o.subscribeOn(Schedulers.io())// io線程處理
            .unsubscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())// 主線程處理
            .subscribe(onNext, onError, onComplete, onSubscribe);
}

使用

初始化

HttpManager.getInstance().init(new RequestInfo(), mServerUrl);

使用MVP或者MVVM的架構，在P或者VM層調用請求

HttpManager.getInstance().request(new PBAppInfo.Builder().build(), 1001, PBResponse.class, new ApiCallback<PBResponse>() {

    @Override
    public void onSuccess(PBResponse response) {
        super.onSuccess(response);
    }

    @Override
    public void onFailure(int code, String message) {
        super.onFailure(code, message);
    }
});