如何订阅和取消订阅或取消 rxjava observable

Question

我是 RxJava 的新手，正在尝试将我的 asyncTask 工作更新到 RxJava。作为第一次尝试，我完成了以下代码：

    public class MainActivity extends AppCompatActivity 
    {
        @Override
        protected void onCreate(Bundle savedInstanceState) 
        {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_main);

            doSomeWork();
        }


        private String funcCallServerGet()
        {
            //Some code to call a HttpClient Get method & return a response string
            //this is the method which previously i used to call inside asynctask doInbackground method
        }


          private void doSomeWork() {
               getSingleObservable()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(getSingleObserver())    ;
        }


        private Single<String> getSingleObservable() 
        {
            return Single.create(new SingleOnSubscribe<String>() {
                @Override
                public void subscribe(SingleEmitter<String> emitter) throws Exception {
                    if(!emitter.isDisposed()) {
                        String strRxResponse =  funcCallServerGet();
                        emitter.onSuccess(strRxResponse);
                    }
                }
            });
        }


         private SingleObserver<String> getSingleObserver() 
        {

            return new SingleObserver<String>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, " onSubscribe getSingleObserver: " + d.isDisposed());             }

                @Override
                public void onSuccess(String value) {
                    Log.d(TAG, " onNext : value : " + value);          }

                @Override
                public void onError(Throwable e) {
                    Log.d(TAG, " onError : " + e.getMessage());         }
            };
        }

    }

但我有一些困惑：

1. 为什么我在 SingleObserver getSingleObserver() 的 onSubscribe() 中得到 false。

2. 在调用活动 onStop() 时如何取消订阅或取消 observable/observer。

3. 另外，屏幕定向时会发生什么。 observable 是自动取消订阅还是继续工作？设备旋转怎么办？

Answer 1

为什么我在 SingleObserver getSingleObserver() 的 onSubscribe() 中出现错误。

您当前正在记录一次性用品是否已在 onSubscribe 方法中处置。此时，一次性用品尚未处理完毕。

当调用活动 onStop() 时，我如何取消订阅或取消 observable/observer。

您可以使用返回一次性的订阅方法，而不是使用 SingleObserver。有了这个，您可以直接管理一次性用品或使用 CompositeDisposable。然后，您将在该一次性用品上调用 dispose 方法，使用 CompositeDisposable 这是通过调用 clear()

来实现的

private final CompositeDisposable disposables = new CompositeDisposable();

@Override
protected void onStart() {
    super.onStart();
    disposables.add(getSingleObservable()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(value -> {
                Log.d(TAG, " onSuccess: " + value);
            }, error -> {
                Log.e(TAG, " onError", error);
            }
        )
    );
}

@Override
protected void onStop() {
    disposables.clear();
    super.onStop();
}

此外，屏幕定向时真正发生的情况。 observable 是自动取消订阅还是继续工作？设备旋转怎么办？

默认情况下不会自动管理 observable，你有责任管理它。在您的示例代码中，当设备旋转时，您将收到对 onCreate 的另一个调用，在这里您正在安排要再次执行的工作，在旋转之前安排的工作仍然可以运行，因此您最终可能会泄漏旧活动并接收工作成功或失败时的回调 - 在这种情况下，您会看到一条日志语句。

有一些工具可以提供自动可观察管理，但您应该阅读作者的文章，了解这种方法存在的一些问题。

• https://blog.danlew.net/2017/08/02/why-not-rxlifecycle/
• https://github.com/trello/RxLifecycle
• https://github.com/uber/AutoDispose

您的另一个选择是查看新的架构组件库，特别是 ViewModel 和 LiveData。这将简化您在订阅管理和配置更改方面需要做的事情。