Android Studio中Java代码：将AsyncTask转换为RxJava求助

第一步：添加依赖 #

在你的build.gradle(Module level)中添加RxJava和RxAndroid的依赖：

implementation "io.reactivex.rxjava3:rxjava:3.1.5"
implementation "io.reactivex.rxjava3:rxandroid:3.0.2"

第二步：替换后的代码实现 #

保留原有downloadUrl()和DirectionsJSONParser逻辑，将两个AsyncTask替换为以下RxJava代码：

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;

// 在Activity/Fragment类中声明Disposable，用于管理订阅生命周期
private Disposable downloadDisposable;

// 执行下载+解析的方法，替代原来的new DownloadTask().execute(url)
private void startDownloadAndParse(String url) {
    downloadDisposable = Observable.fromCallable(() -> {
            // 对应原DownloadTask的doInBackground：后台线程执行下载
            String data = "";
            try {
                data = downloadUrl(url);
                Log.d("DownloadTask", "DownloadTask : " + data);
            } catch (Exception e) {
                Log.d("Background Task", e.toString());
                throw e; // 抛出异常，交由订阅者处理
            }
            return data;
        })
        .subscribeOn(Schedulers.io()) // 指定下载操作在IO线程执行
        .flatMap(jsonData -> Observable.fromCallable(() -> {
            // 对应原ParserTask的doInBackground：后台线程解析JSON
            JSONObject jObject;
            List<List<HashMap<String, String>>> routes = null;
            try {
                jObject = new JSONObject(jsonData);
                DirectionsJSONParser parser = new DirectionsJSONParser();
                routes = parser.parse(jObject);
            } catch (Exception e) {
                e.printStackTrace();
                throw e;
            }
            return routes;
        }))
        .observeOn(AndroidSchedulers.mainThread()) // 指定回调在主线程执行
        .subscribe(
            routes -> {
                // 在这里处理解析完成后的routes数据，对应原AsyncTask执行完成后的逻辑
                // 比如更新UI、使用路由数据等
            },
            error -> {
                // 处理下载或解析过程中的错误
                Log.e("RxTaskError", "任务执行失败: ", error);
            }
        );
}

// 在页面销毁时取消订阅，防止内存泄漏
@Override
protected void onDestroy() {
    super.onDestroy();
    if (downloadDisposable != null && !downloadDisposable.isDisposed()) {
        downloadDisposable.dispose();
    }
}

代码说明 #

• Observable.fromCallable()：将同步耗时操作（下载、解析）包装为Observable流
• subscribeOn(Schedulers.io())：指定上游下载、解析操作在IO线程执行，对应AsyncTask的doInBackground
• observeOn(AndroidSchedulers.mainThread())：指定下游结果回调在主线程执行，对应AsyncTask的onPostExecute
• Disposable：管理订阅生命周期，页面销毁时取消订阅避免内存泄漏
• 链式调用：将下载、解析两个步骤串联，替代原两个AsyncTask依次执行的逻辑

内容的提问来源于stack exchange，提问作者Sam