Retrofit+Rxjava下載文件進(jìn)度的實(shí)現(xiàn)

更新時(shí)間：2017年11月20日 17:08:39 作者：柳擊歌

這篇文章主要介紹了Retrofit+Rxjava下載文件進(jìn)度的實(shí)現(xiàn)，非常不錯(cuò)，具有參考借鑒價(jià)值,需要的朋友可以參考下

前言

最近在學(xué)習(xí)Retrofit，雖然Retrofit沒(méi)有提供文件下載進(jìn)度的回調(diào)，但是Retrofit底層依賴的是OkHttp，實(shí)際上所需要的實(shí)現(xiàn)OkHttp對(duì)下載進(jìn)度的監(jiān)聽(tīng)，在OkHttp的官方Demo中，有一個(gè)Progress.java的文件，顧名思義。點(diǎn)我查看。

準(zhǔn)備工作

本文采用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默認(rèn)的ResponseBody因?yàn)椴恢肋M(jìn)度的相關(guān)信息，所以需要對(duì)其進(jìn)行改造?？梢允褂媒涌诒O(jiān)聽(tīng)進(jìn)度信息。這里采用的是RxBus發(fā)送FileLoadEvent對(duì)象實(shí)現(xiàn)對(duì)下載進(jìn)度的實(shí)時(shí)更新。這里先講改造的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;
  //實(shí)時(shí)發(fā)送當(dāng)前已讀取的字節(jié)和總字節(jié)
  RxBus.getInstance().post(new FileLoadEvent(contentLength(), bytesReaded));
  return bytesRead;
  }
 };
 }
}

呃，OKIO相關(guān)知識(shí)我也正在學(xué)，這個(gè)是從官方Demo中copy的代碼，只不過(guò)中間使用了RxBus實(shí)時(shí)發(fā)送FileLoadEvent對(duì)象。

FileLoadEvent

FileLoadEvent很簡(jiǎn)單，包含了當(dāng)前已加載進(jìn)度和文件總大小。

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 名字看起來(lái)像一個(gè)庫(kù)，但它并不是一個(gè)庫(kù)，而是一種模式，它的思想是使用 RxJava 來(lái)實(shí)現(xiàn)了 EventBus ，而讓你不再需要使用OTTO或者 EventBus。點(diǎn)我查看詳情。

public class RxBus {
 private static volatile RxBus mInstance;
 private SerializedSubject<Object, Object> mSubject;
 private HashMap<String, CompositeSubscription> mSubscriptionMap;
 /**
 * PublishSubject只會(huì)把在訂閱發(fā)生的時(shí)間點(diǎn)之后來(lái)自原始Observable的數(shù)據(jù)發(fā)射給觀察者
 * Subject同時(shí)充當(dāng)了Observer和Observable的角色，Subject是非線程安全的，要避免該問(wèn)題，
 * 需要將 Subject轉(zhuǎn)換為一個(gè) 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ā)送一個(gè)新的事件
 * @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

那么，重點(diǎn)來(lái)了。代碼其實(shí)有5個(gè)方法需要重寫，好吧，其實(shí)這些方法可以精簡(jiǎn)一下。其中progress()方法有兩個(gè)參數(shù)，progress和total,分別表示文件已下載的大小和總大小，我們將這兩個(gè)參數(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());
  }
 }
 }
 /**
 * 訂閱加載的進(jìn)度條
 */
 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 對(duì)異常的處理
  }
 });
 RxBus.getInstance().addSubscription(this, subscription);
 }
 /**
 * 取消訂閱，防止內(nèi)存泄漏
 */
 public void unsubscribe() {
 RxBus.getInstance().unSubscribe(this);
 }
}

開(kāi)始下載

使用自己的ProgressResponseBody

通過(guò)OkHttpClient的攔截器去攔截Response，并將我們的ProgressReponseBody設(shè)置進(jìn)去監(jiān)聽(tīng)進(jì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);
 }
}

請(qǐng)求接口

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

執(zhí)行請(qǐng)求

public void load(String url, final FileCallBack<ResponseBody> callBack){
 apiInfo.download(url)
  .subscribeOn(Schedulers.io())//請(qǐng)求網(wǎng)絡(luò) 在調(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)絡(luò)請(qǐng)求。

通過(guò)V層依賴注入的presenter對(duì)象調(diào)用請(qǐng)求網(wǎng)絡(luò)，請(qǐng)求網(wǎng)絡(luò)后調(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: 對(duì)異常的一些處理
  e.printStackTrace();
  }
 };
 apiManager.load(url, callBack);
 }

踩到的坑。

依賴的Retrofit版本一定要保持一致！??！說(shuō)多了都是淚啊。

保存文件時(shí)要使用RxJava的doOnNext操作符，后續(xù)更新UI的操作切換到UI線程。

總結(jié)

看似代碼很多，其實(shí)過(guò)程并不復(fù)雜：

在保存文件時(shí)，調(diào)用ForwardingSource的read方法，通過(guò)RxBus發(fā)送實(shí)時(shí)的FileLoadEvent對(duì)象。

FileCallBack訂閱RxBus發(fā)送的FileLoadEvent。通過(guò)接收到FileLoadEvent中的下載進(jìn)度和文件總大小對(duì)UI進(jìn)行更新。

在下載保存文件完成后，取消訂閱，防止內(nèi)存泄漏。

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

您可能感興趣的文章: