public class ApkResponseBody extends ResponseBody { 
 
 private Response originalResponse;//原responsebody 
 
 public ApkResponseBody(Response originalResponse) { 
  this.originalResponse = originalResponse; 
 } 
 
 /** 
  * 返回內(nèi)容類型 
  * 
  * @return 
  */ 
 @Override 
 public MediaType contentType() { 
  return originalResponse.body().contentType(); 
 } 
 
 /** 
  * 內(nèi)容總長度 
  * @return 
  */ 
 @Override 
 public long contentLength() { 
  return originalResponse.body().contentLength(); 
 } 
 
 /** 
  * 返回緩存源,類似于io中的BufferedReader 
  * 
  * @return 
  */ 
 @Override 
 public BufferedSource source() { 
 
  return Okio.buffer(new ForwardingSource(originalResponse.body().source()) { 
   long totalRead = 0; 
 
   //返回讀取的長度 
   @Override 
   public long read(Buffer sink, long byteCount) throws IOException { 
    long bytesRead = super.read(sink, byteCount); 
    totalRead += bytesRead == -1 ? 0 : bytesRead; 
    Log.i("test", "本次下載：" + bytesRead); 
    Log.i("test", "總共下載：" + totalRead); 
    RxBus.getDefault().post(new ApkLoadingBean(contentLength(), totalRead)); 
    return bytesRead; 
   } 
 
   @Override 
   public Timeout timeout() { 
    return super.timeout(); 
   } 
 
   @Override 
   public void close() throws IOException { 
    super.close(); 
   } 
 
   @Override 
   public String toString() { 
    return super.toString(); 
   } 
  }); 
 } 
} 

OkHttpClient client = new OkHttpClient().newBuilder() 
    .addInterceptor(new Interceptor() { 
     @Override 
     public Response intercept(Chain chain) throws IOException { 
      Response originalResponse = chain.proceed(chain.request()); 
      return originalResponse 
        .newBuilder() 
        .body(new ApkResponseBody(originalResponse)) 
        .build(); 
     } 
    }).build(); 

public class RxBus { 
 
 private static volatile RxBus mInstance; 
 
 private final Subject<Object> mBus; 
 
 private RxBus() { 
  this.mBus = PublishSubject.create().toSerialized(); 
 } 
 
 public static RxBus getDefault() { 
  if (mInstance == null) { 
   synchronized (RxBus.class) { 
    if (mInstance == null) { 
     mInstance = Holder.BUS; 
    } 
   } 
  } 
  return mInstance; 
 } 
 
 /** 
  * 發(fā)送一個事件 
  * 
  * @param obj 
  */ 
 public void post(Object obj) { 
  mBus.onNext(obj); 
 } 
 
 /** 
  * 暴露出RxBus的Observable供我們訂閱事件 
  * 
  * @param tClass 
  * @param <T> 
  * @return 
  */ 
 public <T> Observable<T> toObservable(Class<T> tClass) { 
  return mBus.ofType(tClass); 
 } 
 
 private static class Holder { 
  private static final RxBus BUS = new RxBus(); 
 } 
} 

public class UpdateApkService extends IntentService { 
 
 private static Context mContext; 
 public static final String ACTION_DOWNLOAD = "intentservice.ACTION_DOWNLOAD"; 
 public static final String DOWNLOAD_URL = "DOWNLOAD_URL"; 
 public static final String APK_PATH = "APK_PATH"; 
 private CompositeDisposable mCompositeDisposable = new CompositeDisposable(); 
 private NotificationCompat.Builder mBuilder; 
 private NotificationManager mNotificationManager; 
 
 public UpdateApkService() { 
  super("UpdateApkService"); 
 } 
 
 public static void startUpdateService(Context context, String url, String apkPath) { 
  mContext = context; 
  Intent intent = new Intent(context, UpdateApkService.class); 
  intent.setPackage(context.getPackageName()); 
  intent.setAction(ACTION_DOWNLOAD); 
  intent.putExtra(DOWNLOAD_URL, url); 
  intent.putExtra(APK_PATH, apkPath); 
  context.startService(intent); 
 } 
 
 @Override 
 protected void onHandleIntent(@Nullable Intent intent) { 
  if (intent != null) { 
   String action = intent.getAction(); 
   if (ACTION_DOWNLOAD.equals(action)) { 
    T.showShort(mContext,"開始下載..."); 
    mNotificationManager = (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE); 
    mBuilder = new NotificationCompat.Builder(this) 
      .setSmallIcon(R.drawable.ic_launcher) 
      .setContentTitle("下載最新版中") 
      .setProgress(100, 0, false) 
      .setAutoCancel(true); 
    mNotificationManager.notify(0, mBuilder.build()); 
    String url = intent.getStringExtra(DOWNLOAD_URL); 
    String apkPath = intent.getStringExtra(APK_PATH); 
    subscribeEvent();//訂閱下載進(jìn)度 
    UpdateManager.downLoadApk(this, url, apkPath, mCompositeDisposable); 
   } 
  } 
 } 
 
 private void subscribeEvent() { 
  RxBus.getDefault().toObservable(ApkLoadingBean.class) 
    .subscribe(new Observer<ApkLoadingBean>() { 
     @Override 
     public void onSubscribe(Disposable d) { 
      mCompositeDisposable.add(d); 
     } 
 
     @Override 
     public void onNext(ApkLoadingBean bean) { 
      int progress = (int) Math.round(bean.getProgress() / (double) bean.getTotal() * 100); 
      mBuilder.setProgress(100, progress, false); 
      mNotificationManager.notify(0, mBuilder.build()); 
      if (progress==100) 
       mNotificationManager.cancel(0); 
     } 
 
     @Override 
     public void onError(Throwable e) { 
      subscribeEvent(); 
     } 
 
     @Override 
     public void onComplete() { 
      subscribeEvent(); 
     } 
    }); 
 } 
 
 @Override 
 public void onDestroy() { 
  super.onDestroy(); 
  Log.i("test", "UpdateService is destory"); 
 } 
}