前言

最近在學習Retrofit忘巧，雖然Retrofit沒有提供文件下載進度的回調(diào)紫皇，但是Retrofit底層依賴的是OkHttp，實際上所需要的實現(xiàn)OkHttp對下載進度的監(jiān)聽，在OkHttp的官方Demo中祝懂，有一個Progress.java的文件，顧名思義拘鞋。點我查看砚蓬。

準備工作

本文采用Dagger2，Retrofit盆色，RxJava灰蛙。

compile'com.squareup.retrofit2:retrofit:2.0.2'
compile'com.squareup.retrofit2:converter-gson:2.0.2'
compile'com.squareup.retrofit2:adapter-rxjava:2.0.2'
//dagger2
compile'com.google.dagger:dagger:2.6'
apt'com.google.dagger:dagger-compiler:2.6'
//RxJava
compile'io.reactivex:rxandroid:1.2.0'
compile'io.reactivex:rxjava:1.1.5'
compile'com.jakewharton.rxbinding:rxbinding:0.4.0'

改造ResponseBody

okHttp3默認的ResponseBody因為不知道進度的相關信息，所以需要對其進行改造隔躲∧ξ啵可以使用接口監(jiān)聽進度信息。這里采用的是RxBus發(fā)送FileLoadEvent對象實現(xiàn)對下載進度的實時更新蹭越。這里先講改造的ProgressResponseBody障本。

public class ProgressResponseBody extends ResponseBody {
    private ResponseBody responseBody;

    private BufferedSource bufferedSource;
    public ProgressResponseBody(ResponseBody responseBody) {
        this.responseBody = responseBody;
    }

    @Override
    public MediaType contentType() {
        return responseBody.contentType();
    }

    @Override
    public long contentLength() {
        return responseBody.contentLength();
    }

    @Override
    public BufferedSource source() {
        if (bufferedSource == null) {
            bufferedSource = Okio.buffer(source(responseBody.source()));
        }
        return bufferedSource;
    }

    private Source source(Source source) {
        return new ForwardingSource(source) {
            long bytesReaded = 0;
            @Override
            public long read(Buffer sink, long byteCount) throws IOException {
                long bytesRead = super.read(sink, byteCount);
                bytesReaded += bytesRead == -1 ? 0 : bytesRead;
               //實時發(fā)送當前已讀取的字節(jié)和總字節(jié)
                RxBus.getInstance().post(new FileLoadEvent(contentLength(), bytesReaded));
                return bytesRead;
            }
        };
    }
}

呃，OKIO相關知識我也正在學响鹃，這個是從官方Demo中copy的代碼驾霜，只不過中間使用了RxBus實時發(fā)送FileLoadEvent對象。

FileLoadEvent

FileLoadEvent很簡單买置，包含了當前已加載進度和文件總大小粪糙。

public class FileLoadEvent {

    long total;
    long bytesLoaded;

    public long getBytesLoaded() {
        return bytesLoaded;
    }

    public long getTotal() {
        return total;
    }

    public FileLoadEvent(long total, long bytesLoaded) {
        this.total = total;
        this.bytesLoaded = bytesLoaded;
    }
}

RxBus

RxBus 名字看起來像一個庫，但它并不是一個庫忿项，而是一種模式蓉冈，它的思想是使用 RxJava 來實現(xiàn)了 EventBus ，而讓你不再需要使用OTTO或者 EventBus轩触。點我查看詳情寞酿。

public class RxBus {

    private static volatile RxBus mInstance;
    private SerializedSubject<Object, Object> mSubject;
    private HashMap<String, CompositeSubscription> mSubscriptionMap;

    /**
     *  PublishSubject只會把在訂閱發(fā)生的時間點之后來自原始Observable的數(shù)據(jù)發(fā)射給觀察者
     *  Subject同時充當了Observer和Observable的角色，Subject是非線程安全的脱柱，要避免該問題伐弹，
     *  需要將 Subject轉(zhuǎn)換為一個 SerializedSubject ，上述RxBus類中把線程非安全的PublishSubject包裝成線程安全的Subject榨为。
     */
    private RxBus() {
        mSubject = new SerializedSubject<>(PublishSubject.create());
    }

    /**
     * 單例 雙重鎖
     * @return
     */
    public static RxBus getInstance() {
        if (mInstance == null) {
            synchronized (RxBus.class) {
                if (mInstance == null) {
                    mInstance = new RxBus();
                }
            }
        }
        return mInstance;
    }

    /**
     * 發(fā)送一個新的事件
     * @param o
     */
    public void post(Object o) {
        mSubject.onNext(o);
    }

    /**
     * 根據(jù)傳遞的 eventType 類型返回特定類型(eventType)的 被觀察者
     * @param type
     * @param <T>
     * @return
     */
    public <T> Observable<T> tObservable(final Class<T> type) {
        //ofType操作符只發(fā)射指定類型的數(shù)據(jù)惨好，其內(nèi)部就是filter+cast
        return mSubject.ofType(type);
    }

    public <T> Subscription doSubscribe(Class<T> type, Action1<T> next, Action1<Throwable> error) {
        return tObservable(type)
                .onBackpressureBuffer()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next, error);
    }

    public void addSubscription(Object o, Subscription subscription) {

        if (mSubscriptionMap == null) {
            mSubscriptionMap = new HashMap<>();
        }
        String key = o.getClass().getName();
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).add(subscription);
        } else {
            CompositeSubscription compositeSubscription = new CompositeSubscription();
            compositeSubscription.add(subscription);
            mSubscriptionMap.put(key, compositeSubscription);
          //  Log.e("air", "addSubscription:訂閱成功 " );
        }
    }

    public void unSubscribe(Object o) {
        if (mSubscriptionMap == null) {
            return;
        }
        String key = o.getClass().getName();
        if (!mSubscriptionMap.containsKey(key)) {
            return;
        }
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).unsubscribe();
        }
        mSubscriptionMap.remove(key);
        //Log.e("air", "unSubscribe: 取消訂閱" );
    }
}

FileCallBack

那么，重點來了随闺。代碼其實有5個方法需要重寫日川，好吧，其實這些方法可以精簡一下矩乐。其中progress()方法有兩個參數(shù)龄句，progress和total,分別表示文件已下載的大小和總大小，我們將這兩個參數(shù)不斷更新到UI上就行了。

public abstract class FileCallBack<T> {

    private String destFileDir;
    private String destFileName;

    public FileCallBack(String destFileDir, String destFileName) {
        this.destFileDir = destFileDir;
        this.destFileName = destFileName;
        subscribeLoadProgress();
    }

    public abstract void onSuccess(T t);

    public abstract void progress(long progress, long total);

    public abstract void onStart();

    public abstract void onCompleted();

    public abstract void onError(Throwable e);

    public void saveFile(ResponseBody body) {
        InputStream is = null;
        byte[] buf = new byte[2048];
        int len;
        FileOutputStream fos = null;
        try {
            is = body.byteStream();
            File dir = new File(destFileDir);
            if (!dir.exists()) {
                dir.mkdirs();
            }
            File file = new File(dir, destFileName);
            fos = new FileOutputStream(file);
            while ((len = is.read(buf)) != -1) {
                fos.write(buf, 0, len);
            }
            fos.flush();
            unsubscribe();
            //onCompleted();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (is != null) is.close();
                if (fos != null) fos.close();
            } catch (IOException e) {
                Log.e("saveFile", e.getMessage());
            }
        }
    }

    /**
     * 訂閱加載的進度條
     */
    public void subscribeLoadProgress() {
        Subscription subscription = RxBus.getInstance().doSubscribe(FileLoadEvent.class, new Action1<FileLoadEvent>() {
            @Override
            public void call(FileLoadEvent fileLoadEvent) {
                progress(fileLoadEvent.getBytesLoaded(),fileLoadEvent.getTotal());
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                //TODO 對異常的處理
            }
        });
        RxBus.getInstance().addSubscription(this, subscription);
    }

    /**
     * 取消訂閱撒璧，防止內(nèi)存泄漏
     */
    public void unsubscribe() {
        RxBus.getInstance().unSubscribe(this);
    }

}

開始下載

使用自己的ProgressResponseBody

通過OkHttpClient的攔截器去攔截Response透葛，并將我們的ProgressReponseBody設置進去監(jiān)聽進度。

public class ProgressInterceptor implements Interceptor {
    @Override
    public Response intercept(Chain chain) throws IOException {
        Response originalResponse = chain.proceed(chain.request());
        return originalResponse.newBuilder()
                .body(new ProgressResponseBody(originalResponse.body()))
                .build();
    }
}

構(gòu)建Retrofit

@Module
public class ApiModule {

    @Provides
    @Singleton
    public OkHttpClient provideClient() {
        OkHttpClient client = new OkHttpClient.Builder()
                .addInterceptor(new ProgressInterceptor())
                .build();
        return client;
    }

    @Provides
    @Singleton
    public Retrofit provideRetrofit(OkHttpClient client){
        Retrofit retrofit = new Retrofit.Builder()
                .client(client)
                .baseUrl(Constant.HOST)
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .build();
        return retrofit;
    }

    @Provides
    @Singleton
    public ApiInfo provideApiInfo(Retrofit retrofit){
        return retrofit.create(ApiInfo.class);
    }

    @Provides
    @Singleton
    public ApiManager provideApiManager(Application application, ApiInfo apiInfo){
        return new ApiManager(application,apiInfo);
    }

}

請求接口

public interface ApiInfo {
    @Streaming
    @GET
    Observable<ResponseBody> download(@Url String url);
}

執(zhí)行請求

public void load(String url, final FileCallBack<ResponseBody> callBack){
        apiInfo.download(url)
                .subscribeOn(Schedulers.io())//請求網(wǎng)絡 在調(diào)度者的io線程
                .observeOn(Schedulers.io()) //指定線程保存文件
                .doOnNext(new Action1<ResponseBody>() {
                    @Override
                    public void call(ResponseBody body) {
                        callBack.saveFile(body);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread()) //在主線程中更新ui
                .subscribe(new FileSubscriber<ResponseBody>(application,callBack));
    }

在presenter層中執(zhí)行網(wǎng)絡請求卿樱。

通過V層依賴注入的presenter對象調(diào)用請求網(wǎng)絡，請求網(wǎng)絡后調(diào)用V層更新UI的操作硫椰。

public void load(String url){
        String fileName = "app.apk";
        String fileStoreDir = Environment.getExternalStorageDirectory().getAbsolutePath();
        Log.e(TAG, "load: "+fileStoreDir.toString() );
        FileCallBack<ResponseBody> callBack = new FileCallBack<ResponseBody>(fileStoreDir,fileName) {

            @Override
            public void onSuccess(final ResponseBody responseBody) {
                Toast.makeText(App.getInstance(),"下載文件成功",Toast.LENGTH_SHORT).show();
            }

            @Override
            public void progress(long progress, long total) {
                iHomeView.update(total,progress);
            }

            @Override
            public void onStart() {
                iHomeView.showLoading();
            }

            @Override
            public void onCompleted() {
                iHomeView.hideLoading();
            }

            @Override
            public void onError(Throwable e) {
                //TODO: 對異常的一些處理
                e.printStackTrace();
            }
        };
        apiManager.load(url, callBack);
    }

踩到的坑繁调。

• 依賴的Retrofit版本一定要保持一致！０胁荨蹄胰！說多了都是淚啊。
• 保存文件時要使用RxJava的doOnNext操作符奕翔，后續(xù)更新UI的操作切換到UI線程裕寨。

總結(jié)

看似代碼很多，其實過程并不復雜：

• 在保存文件時派继，調(diào)用ForwardingSource的read方法宾袜，通過RxBus發(fā)送實時的FileLoadEvent對象。
• FileCallBack訂閱RxBus發(fā)送的FileLoadEvent驾窟。通過接收到FileLoadEvent中的下載進度和文件總大小對UI進行更新庆猫。
• 在下載保存文件完成后，取消訂閱绅络，防止內(nèi)存泄漏月培。

Demo地址：https://github.com/AirMiya/DownloadDemo