0

我正在使用rxjava3并且不太理解为什么添加ObserveOnand后方法没有在流中被调用SubscribeOn。

这是示例 java 代码：

package mytestapp.error;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.json.JSONTokener;

/**
 * Hello world!
 *
 */
public class App {
    public static void main(String[] args) {
        System.out.println("Hello World!");
        String apiUrl = "myApiUrl";
        try {
            App app = new App();
            app.syncNow(apiUrl);
        } catch (JSONException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    private void syncNow(String apiUrl) throws JSONException, IOException {
        createOrAlterTable().observeOn(Schedulers.newThread())
                .switchMap(d -> menuTableRecords(apiUrl))
                .observeOn(Schedulers.newThread()).subscribe(res -> {
                    System.out.println(res);

                }, onError -> {
                    System.out.println(onError);
                }, () -> {
                    System.out.println("Completed!!");
                });
        ;
    }

    private @NonNull Observable<Object> createOrAlterTable()
            throws IOException, JSONException {
        // Read table from backend
        // Read last sync file
        // get user data
        return Observable.zip(readTableFromBackend(), readLastSyncFile(),
                getUserData(),
                (s1, s2, s3) -> readTableFromBackendZipperFun(s1, s2, s3)).subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io()).map(
                d -> d);
    }

    private @NonNull Observable<String> readTableFromBackend()
            throws JSONException, IOException {
        return Observable.fromArray("testing");
    }

    private @NonNull Observable<JSONObject> readLastSyncFile()
            throws JSONException {
        return Observable.fromArray(new JSONObject());
    }

    private @NonNull Observable<Boolean> getUserData() throws JSONException {
        return Observable.fromArray(true);
    }

    private JSONArray readTableFromBackendZipperFun(String sqlliteDDL,
            JSONObject lastFV, boolean userDataFlag) throws JSONException {
        System.out.println("zip ops");
        return new JSONArray();
    }

    private @NonNull Observable<String> menuTableRecords(String apiUrl)
            throws JSONException, IOException {
        
        return Observable.fromArray("MENU_TABLE_RECORDS");
    }
}

我想在单独的线程上执行每个方法，也想在不同的线程上执行订阅。

造成问题的原因是什么。如何解决上述情况？

谢谢。

4

1 回答 1

1

看起来您无需等待异步活动在该main方法中完成，您的应用程序就会退出。推荐阅读：https ://github.com/ReactiveX/RxJava#simple-background-computation

您可以通过使用blockingSubscribe而不是仅subscribe在这种特殊情况下解决问题。

于 2020-09-11T09:50:40.200 回答