/*斷點(diǎn)續(xù)傳下載接口*/  @Streaming/*大文件需要加入這個(gè)判斷，防止下載過程中寫入到內(nèi)存中*/  @GET  Observable<ResponseBody> download(@Header("RANGE") String start, @Url String url);

/** * 自定義進(jìn)度的body * @author wzg */public class DownloadResponseBody extends ResponseBody {  private ResponseBody responseBody;  private DownloadProgressListener progressListener;  private BufferedSource bufferedSource;  public DownloadResponseBody(ResponseBody responseBody, DownloadProgressListener progressListener) {    this.responseBody = responseBody;    this.progressListener = progressListener;  }  @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 totalBytesRead = 0L;      @Override      public long read(Buffer sink, long byteCount) throws IOException {        long bytesRead = super.read(sink, byteCount);        // read() returns the number of bytes read, or -1 if this source is exhausted.        totalBytesRead += bytesRead != -1 ? bytesRead : 0;        if (null != progressListener) {          progressListener.update(totalBytesRead, responseBody.contentLength(), bytesRead == -1);        }        return bytesRead;      }    };  }}

/** * 成功回調(diào)處理 * Created by WZG on 2016/10/20. */public interface DownloadProgressListener {  /**   * 下載進(jìn)度   * @param read   * @param count   * @param done   */  void update(long read, long count, boolean done);}

/** * 成功回調(diào)處理 * Created by WZG on 2016/10/20. */public class DownloadInterceptor implements Interceptor {  private DownloadProgressListener listener;  public DownloadInterceptor(DownloadProgressListener listener) {    this.listener = listener;  }  @Override  public Response intercept(Chain chain) throws IOException {    Response originalResponse = chain.proceed(chain.request());    return originalResponse.newBuilder()        .body(new DownloadResponseBody(originalResponse.body(), listener))        .build();  }}

public class DownInfo {  /*存儲(chǔ)位置*/  private String savePath;  /*下載url*/  private String url;  /*基礎(chǔ)url*/  private String baseUrl;  /*文件總長(zhǎng)度*/  private long countLength;  /*下載長(zhǎng)度*/  private long readLength;  /*下載唯一的HttpService*/  private HttpService service;  /*回調(diào)監(jiān)聽*/  private HttpProgressOnNextListener listener;  /*超時(shí)設(shè)置*/  private int DEFAULT_TIMEOUT = 6;  /*下載狀態(tài)*/  private DownState state;  }

public enum DownState {  START,  DOWN,  PAUSE,  STOP,  ERROR,  FINISH,}

/** * 下載過程中的回調(diào)處理 * Created by WZG on 2016/10/20. */public abstract class HttpProgressOnNextListener<T> {  /**   * 成功后回調(diào)方法   * @param t   */  public abstract void onNext(T t);  /**   * 開始下載   */  public abstract void onStart();  /**   * 完成下載   */  public abstract void onComplete();  /**   * 下載進(jìn)度   * @param readLength   * @param countLength   */  public abstract void updateProgress(long readLength, long countLength);  /**   * 失敗或者錯(cuò)誤方法   * 主動(dòng)調(diào)用，更加靈活   * @param e   */   public void onError(Throwable e){   }  /**   * 暫停下載   */  public void onPuase(){  }  /**   * 停止下載銷毀   */  public void onStop(){  }}

/** * 用于在Http請(qǐng)求開始時(shí)，自動(dòng)顯示一個(gè)ProgressDialog * 在Http請(qǐng)求結(jié)束是，關(guān)閉ProgressDialog * 調(diào)用者自己對(duì)請(qǐng)求數(shù)據(jù)進(jìn)行處理 * Created by WZG on 2016/7/16. */public class ProgressDownSubscriber<T> extends Subscriber<T> implements DownloadProgressListener {  //弱引用結(jié)果回調(diào)  private WeakReference<HttpProgressOnNextListener> mSubscriberOnNextListener;  /*下載數(shù)據(jù)*/  private DownInfo downInfo;  public ProgressDownSubscriber(DownInfo downInfo) {    this.mSubscriberOnNextListener = new WeakReference<>(downInfo.getListener());    this.downInfo=downInfo;  }  /**   * 訂閱開始時(shí)調(diào)用   * 顯示ProgressDialog   */  @Override  public void onStart() {    if(mSubscriberOnNextListener.get()!=null){      mSubscriberOnNextListener.get().onStart();    }    downInfo.setState(DownState.START);  }  /**   * 完成，隱藏ProgressDialog   */  @Override  public void onCompleted() {    if(mSubscriberOnNextListener.get()!=null){      mSubscriberOnNextListener.get().onComplete();    }    downInfo.setState(DownState.FINISH);  }  /**   * 對(duì)錯(cuò)誤進(jìn)行統(tǒng)一處理   * 隱藏ProgressDialog   *   * @param e   */  @Override  public void onError(Throwable e) {    /*停止下載*/    HttpDownManager.getInstance().stopDown(downInfo);    if(mSubscriberOnNextListener.get()!=null){      mSubscriberOnNextListener.get().onError(e);    }    downInfo.setState(DownState.ERROR);  }  /**   * 將onNext方法中的返回結(jié)果交給Activity或Fragment自己處理   *   * @param t 創(chuàng)建Subscriber時(shí)的泛型類型   */  @Override  public void onNext(T t) {    if (mSubscriberOnNextListener.get() != null) {      mSubscriberOnNextListener.get().onNext(t);    }  }  @Override  public void update(long read, long count, boolean done) {    if(downInfo.getCountLength()>count){      read=downInfo.getCountLength()-count+read;    }else{      downInfo.setCountLength(count);    }    downInfo.setReadLength(read);    if (mSubscriberOnNextListener.get() != null) {      /*接受進(jìn)度消息，造成UI阻塞，如果不需要顯示進(jìn)度可去掉實(shí)現(xiàn)邏輯，減少壓力*/      rx.Observable.just(read).observeOn(AndroidSchedulers.mainThread())          .subscribe(new Action1<Long>() {        @Override        public void call(Long aLong) {           /*如果暫?；蛘咄Ｖ?fàn)顟B(tài)延遲，不需要繼續(xù)發(fā)送回調(diào)，影響顯示*/          if(downInfo.getState()==DownState.PAUSE||downInfo.getState()==DownState.STOP)return;          downInfo.setState(DownState.DOWN);          mSubscriberOnNextListener.get().updateProgress(aLong,downInfo.getCountLength());        }      });    }  }}

 /**   * 獲取單例   * @return   */  public static HttpDownManager getInstance() {    if (INSTANCE == null) {      synchronized (HttpDownManager.class) {        if (INSTANCE == null) {          INSTANCE = new HttpDownManager();        }      }    }    return INSTANCE;  }

 /*回調(diào)sub隊(duì)列*/  private HashMap<String,ProgressDownSubscriber> subMap;  /*單利對(duì)象*/  private volatile static HttpDownManager INSTANCE;  private HttpDownManager(){    downInfos=new HashSet<>();    subMap=new HashMap<>();  }

 /**   * 開始下載   */  public void startDown(DownInfo info){    /*正在下載不處理*/    if(info==null||subMap.get(info.getUrl())!=null){      return;    }    /*添加回調(diào)處理類*/    ProgressDownSubscriber subscriber=new ProgressDownSubscriber(info);    /*記錄回調(diào)sub*/    subMap.put(info.getUrl(),subscriber);    /*獲取service，多次請(qǐng)求公用一個(gè)sercie*/    HttpService httpService;    if(downInfos.contains(info)){      httpService=info.getService();    }else{      DownloadInterceptor interceptor = new DownloadInterceptor(subscriber);      OkHttpClient.Builder builder = new OkHttpClient.Builder();      //手動(dòng)創(chuàng)建一個(gè)OkHttpClient并設(shè)置超時(shí)時(shí)間      builder.connectTimeout(info.getConnectionTime(), TimeUnit.SECONDS);      builder.addInterceptor(interceptor);      Retrofit retrofit = new Retrofit.Builder()          .client(builder.build())          .addConverterFactory(GsonConverterFactory.create())          .addCallAdapterFactory(RxJavaCallAdapterFactory.create())          .baseUrl(info.getBaseUrl())          .build();      httpService= retrofit.create(HttpService.class);      info.setService(httpService);    }    /*得到rx對(duì)象-上一次下載的位置開始下載*/    httpService.download("bytes=" + info.getReadLength() + "-",info.getUrl())        /*指定線程*/        .subscribeOn(Schedulers.io())        .unsubscribeOn(Schedulers.io())          /*失敗后的retry配置*/        .retryWhen(new RetryWhenNetworkException())        /*讀取下載寫入文件*/        .map(new Func1<ResponseBody, DownInfo>() {          @Override          public DownInfo call(ResponseBody responseBody) {            try {              writeCache(responseBody,new File(info.getSavePath()),info);            } catch (IOException e) {              /*失敗拋出異常*/              throw new HttpTimeException(e.getMessage());            }            return info;          }        })        /*回調(diào)線程*/        .observeOn(AndroidSchedulers.mainThread())        /*數(shù)據(jù)回調(diào)*/        .subscribe(subscriber);  }

  /**   * 寫入文件   * @param file   * @param info   * @throws IOException   */  public void writeCache(ResponseBody responseBody,File file,DownInfo info) throws IOException{    if (!file.getParentFile().exists())      file.getParentFile().mkdirs();    long allLength;    if (info.getCountLength()==0){      allLength=responseBody.contentLength();    }else{      allLength=info.getCountLength();    }      FileChannel channelOut = null;      RandomAccessFile randomAccessFile = null;      randomAccessFile = new RandomAccessFile(file, "rwd");      channelOut = randomAccessFile.getChannel();      MappedByteBuffer mappedBuffer = channelOut.map(FileChannel.MapMode.READ_WRITE,          info.getReadLength(),allLength-info.getReadLength());      byte[] buffer = new byte[1024*8];      int len;      int record = 0;      while ((len = responseBody.byteStream().read(buffer)) != -1) {        mappedBuffer.put(buffer, 0, len);        record += len;      }      responseBody.byteStream().close();        if (channelOut != null) {          channelOut.close();        }        if (randomAccessFile != null) {          randomAccessFile.close();        }  }

/**   * 停止下載   */  public void stopDown(DownInfo info){    if(info==null)return;    info.setState(DownState.STOP);    info.getListener().onStop();    if(subMap.containsKey(info.getUrl())) {      ProgressDownSubscriber subscriber=subMap.get(info.getUrl());      subscriber.unsubscribe();      subMap.remove(info.getUrl());    }    /*同步數(shù)據(jù)庫*/  }

 /**   * 暫停下載   * @param info   */  public void pause(DownInfo info){    if(info==null)return;    info.setState(DownState.PAUSE);    info.getListener().onPuase();    if(subMap.containsKey(info.getUrl())){      ProgressDownSubscriber subscriber=subMap.get(info.getUrl());      subscriber.unsubscribe();      subMap.remove(info.getUrl());    }    /*這里需要講info信息寫入到數(shù)據(jù)中，可自由擴(kuò)展，用自己項(xiàng)目的數(shù)據(jù)庫*/  }

/**   * 停止全部下載   */  public void stopAllDown(){    for (DownInfo downInfo : downInfos) {      stopDown(downInfo);    }    subMap.clear();    downInfos.clear();  }  /**   * 暫停全部下載   */  public void pauseAll(){    for (DownInfo downInfo : downInfos) {      pause(downInfo);    }    subMap.clear();    downInfos.clear();  }