RxJava中的AsyncTask

Question

我無法理解如何在RxJava中轉換簡單的AsyncTask。舉例：

private class Sync extends AsyncTask<String, String, String> { 

    @Override 
    protected String doInBackground(String... params) { 
       String proxy_arr = ""; 
        try { 
         Document jsoup_proxy = Jsoup.connect(Constants.SITE_PROXY_LIST) 
           .userAgent(Constants.USER_AGENT) 
           .ignoreContentType(true) 
           .ignoreHttpErrors(true) 
           .timeout(Constants.USER_TIMEOUT) 
           .get(); 

         if (jsoup_proxy != null) proxy_arr = jsoup_proxy.text().trim(); 
        } catch (IOException e) { 
         new DebugLog(getActivity(), "News", "Sync PROXY", Log.getStackTraceString(e)); 
        } 
       return proxy_arr; 
    } 

    @Override 
    protected void onPostExecute(String result) { 
     if (result.equals("err_internet")){ 
      func.toastMessage(R.string.toast_err_nointernet, "", "alert"); 
     } 

     reloadAdapter(); 
    } 
} 

因爲它可以在相同的工作條件下翻譯RxJava？ 謝謝！

Answer 1

而不是使用Observable.create，你應該使用Observable.defer()或更好，但Observable.fromCallable（這是在RxJava 1.0.15中引入） - 因爲這些方法將確保正確觀察的合同，並節省您的一些錯誤，你可以通過創建時觀察到的介紹手。

而不是按照上述答案中的建議與runOnUiThread一起使用，您應該真正使用AndroidSchedulers.mainThread()，它是爲此目的而創建的。只需使用提供它的RxAndroid庫。

我建議以下解決方案：

public Observable<String> getJsoupProxy() { 
    return Observable.fromCallable(() -> { 
     try { 
     Document jsoup_proxy = Jsoup.connect(Constants.SITE_PROXY_LIST) 
      .userAgent(Constants.USER_AGENT) 
      .ignoreContentType(true) 
      .ignoreHttpErrors(true) 
      .timeout(Constants.USER_TIMEOUT) 
      .get(); 

     return jsoup_proxy != null ? jsoup_proxy.text().trim() : ""; 
     } catch (IOException e) { 
     // just rethrow as RuntimeException to be caught in subscriber's onError 
     throw new RuntimeException(e); 
     } 
    }); 
} 

getJsoupProxy() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) // this scheduler is exported by RxAndroid library 
    .subscribe(
    proxy -> { 
     if(proxy.equals("err_internet")) { 
     // toast 
     } 
     reloadAdapter(); 
    }, 
    error -> new DebugLog(getActivity(), "News", "Sync PROXY", Log.getStackTraceString(error))); 

Answer 2

這是一種方法。如果需要的話，你可以放棄延期。

 Observable.defer(new Func0<Observable<String>>() { 
       @Override 
       public Observable<String> call() { 
        return Observable.create(new Observable.OnSubscribe<String>() { 
         @Override 
         public void call(Subscriber<? super String> subscriber) { 
          String proxy_arr = ""; 
          try { 
           Document jsoup_proxy = Jsoup.connect(Constants.SITE_PROXY_LIST) 
             .userAgent(Constants.USER_AGENT) 
             .ignoreContentType(true) 
             .ignoreHttpErrors(true) 
             .timeout(Constants.USER_TIMEOUT) 
             .get(); 

           if (jsoup_proxy != null) proxy_arr = jsoup_proxy.text().trim(); 
          } catch (IOException e) { 
           new DebugLog(getActivity(), "News", "Sync PROXY", Log.getStackTraceString(e)); 
          } 
          if (!subscriber.isUnsubscribed()) { 
           subscriber.onNext(proxy_arr); 
          } 
         } 
        }) 
       } 
      }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Action1<String>() { 
       @Override 
       public void call(String result) { 
        if (result.equals("err_internet")){ 
         func.toastMessage(R.string.toast_err_nointernet, "", "alert"); 
        } 
        reloadAdapter(); 
       } 
      }); 

Answer 3

當轉換功能是被動請記住，你應該定義

• onNext
• 的onError
• onCompleted

「事件」。

實際上rx在數據序列上運行良好，當然你也可以創建一個只有一個發射項的數據序列。 所以要修改你的方法是被動的，我會先說你應該分離責任。

某處在一個倉庫類或你的名字，根據你的架構，你只需要創建這樣的：

public Observable<String> getProxyAsync() { 
     return Observable.create(subscriber -> { 
      String proxy_arr = ""; 
      try { 
       Document jsoup_proxy = Jsoup.connect(Constants.SITE_PROXY_LIST) 
        .userAgent(Constants.USER_AGENT) 
        .ignoreContentType(true) 
        .ignoreHttpErrors(true) 
        .timeout(Constants.USER_TIMEOUT) 
        .get(); 

       if (jsoup_proxy != null) proxy_arr = jsoup_proxy.text().trim(); 

       subscriber.onNext(proxy_arr); 
      } catch (IOException e) { 
       subscriber.onError(e); 
      } finally { 
       subscriber.onCompleted(); 
      } 
     }); 
    } 

後

，靠近活動的地方，只是訂閱此方法是這樣的：

public void myPreciousMethod() { 
    myCustomRepo.getProxyAsync() 
     .subscribeOn(Schedulers.newThread()) 
     .subscribe(result -> { 
      runOnUiThread(() -> { 
       if (result.equals("err_internet")) { 
        func.toastMessage(R.string.toast_err_nointernet, "", "alert"); 
       } 
      }); 
     }, throwable -> { 
      // some exception happened emmited by your code, handle it well 
      new DebugLog(getActivity(), "News", "Sync PROXY", Log.getStackTraceString(e)); 
     },() -> { 
      // onCompleted: 
      runOnUiThread(() -> reloadAdapter()); 
     }); 
} 

我建議使用.runOnUiThread()（在您的活動或任何其他視圖相關的操作與rx），以避免背壓，但它確實取決於發射數據的數量和頻率。 （您也可以使用.observeOn()和.subscribeOn()）爲了更簡潔的代碼，還強烈建議使用retrolambda。