在流方法链中添加 SubscribeOn 和 ObserveOn 后未触发

Question

我正在使用 rxjava3，不太理解为什么在添加 ObserveOn 和 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");
    }
}

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

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

谢谢。

Answer 1

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

在此特定情况下，您可以使用 blockingSubscribe 而不是仅 subscribe 来解决该问题。

在流方法链中添加 SubscribeOn 和 ObserveOn 后未触发

After added SubscribeOn and ObserveOn in stream method chaining not getting trigger

java

rx-java3