Retrofit+Rxjava下載文件進度的示例分析

這篇文章主要為大家展示了“Retrofit+Rxjava下載文件進度的示例分析”，內容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領大家一起研究并學習一下“Retrofit+Rxjava下載文件進度的示例分析”這篇文章吧。

準備工作

本文采用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因為不知道進度的相關信息，所以需要對其進行改造。可以使用接口監聽進度信息。這里采用的是RxBus發送FileLoadEvent對象實現對下載進度的實時更新。這里先講改造的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;
  //實時發送當前已讀取的字節和總字節
  RxBus.getInstance().post(new FileLoadEvent(contentLength(), bytesReaded));
  return bytesRead;
  }
 };
 }
}

呃，OKIO相關知識我也正在學，這個是從官方Demo中copy的代碼，只不過中間使用了RxBus實時發送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 來實現了 EventBus ，而讓你不再需要使用OTTO或者 EventBus。點我查看詳情。

public class RxBus {
 private static volatile RxBus mInstance;
 private SerializedSubject<Object, Object> mSubject;
 private HashMap<String, CompositeSubscription> mSubscriptionMap;
 /**
 * PublishSubject只會把在訂閱發生的時間點之后來自原始Observable的數據發射給觀察者
 * Subject同時充當了Observer和Observable的角色，Subject是非線程安全的，要避免該問題，
 * 需要將 Subject轉換為一個 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;
 }
 /**
 * 發送一個新的事件
 * @param o
 */
 public void post(Object o) {
 mSubject.onNext(o);
 }
 /**
 * 根據傳遞的 eventType 類型返回特定類型(eventType)的 被觀察者
 * @param type
 * @param <T>
 * @return
 */
 public <T> Observable<T> tObservable(final Class<T> type) {
 //ofType操作符只發射指定類型的數據，其內部就是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()方法有兩個參數，progress和total,分別表示文件已下載的大小和總大小，我們將這兩個參數不斷更新到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);
 }
 /**
 * 取消訂閱，防止內存泄漏
 */
 public void unsubscribe() {
 RxBus.getInstance().unSubscribe(this);
 }
}

開始下載

使用自己的ProgressResponseBody

通過OkHttpClient的攔截器去攔截Response，并將我們的ProgressReponseBody設置進去監聽進度。

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();
 }
}

構建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);
}

執行請求

public void load(String url, final FileCallBack<ResponseBody> callBack){
 apiInfo.download(url)
  .subscribeOn(Schedulers.io())//請求網絡 在調度者的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層中執行網絡請求。

通過V層依賴注入的presenter對象調用請求網絡，請求網絡后調用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下載文件進度的示例分析”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！