java - RxJava的Observable订阅时如何作到在onComplete被调用时取消订阅？

Question

例:

table.subscribe(tableIns -> {            // System.out.println("-------Information-------");            System.out.println(tableIns);        }, throwable -> {            throw new SchemaExportException(throwable);        }, new Action() {            @Override            public void run() throws Exception {                System.out.println("Complete");                // 在这里取消订阅        });

注意：是非Android 运行环境, 使用的是RxJava2.x

Answer 1

你可以在 Action 的 run 方法中取消订阅，就像这样：

Observable.create(emitter -> {    // 你的代码...}).subscribe(tableIns -> {    System.out.println(tableIns);}, throwable -> {    throw new SchemaExportException(throwable);}, () -> {    System.out.println("Complete");    emitter.unsubscribe(); // 在这里取消订阅});

这段代码中，我们使用 Observable.create 来创建一个 Observable。这个 Observable 在执行过程中会调用 emitter.onNext，emitter.onError 和 emitter.onComplete。在 onComplete 被调用时，我们调用 emitter.unsubscribe() 来取消订阅。

注意，你需要确保 emitter 在 onComplete 被调用时仍然可用。如果 emitter 在 onComplete 调用之前就已经不再可用，那么你可能会遇到问题。