是否需要退订

Question

我很难理解 RX。以下情况是否需要退订？ "call"函数执行后有没有自动退订的方法？

Observable.create(new Observable.OnSubscribe<NumberInfo>() {
        @Override
        public void call(Subscriber<? super NumberInfo> subscriber) {
            try {
                // Store data to db
            } catch (Exception e) {
                Log.e(TAG, "Downloaded numberInfo was not added to cache.", e);
            }
        }
    }).subscribeOn(Schedulers.newThread())
    .subscribe();

我不想观察任何结果，所以我省略了经典 .observeOn(AndroidSchedulers.mainThread())

感谢解释。

Answer 1

当 Observable 完成时，RxJava 会自动取消订阅。您需要拨打subscriber.onComplete()进行自动退订。

Answer 2

根据 Rx 合约，当 Observable 触发 onCompleted 时，Observer 取消订阅。在您的情况下，合同未得到遵守，因为您的代码中没有 subscriber.onCompleted()。

如果您只需要 "Fire and forget" 之类的东西，您可以尝试：

Schedulers.io().createWorker().schedule(new Action0() {
    @Override
    public void call() {
        try {
            // Store data to db
        } catch (Exception e) {
            Log.e(TAG, "Downloaded numberInfo was not added to cache.", e);
        }
    }
});

它将在 I/O 调度程序上执行并且您的 UI 线程是安全的。

IMO 你应该总是有一个 return 值。您的 Store data to db 路由肯定有一些 return 值，例如 long 指定行号或 boolean 表示成功。使用这种方法，您可以创建一个合适的方法：

public Observable<Long> storeToDb(final SomethingToStore storeMe) {
    return Observable
            .create(new Observable.OnSubscribe<Long>() {
                @Override
                public void call(Subscriber<? super Long> subscriber) {
                    long row = syncStore(storeMe);

                    if (row == -1) {
                        subscriber.onError(new Throwable("Cannot store " + storeMe.toString + " to DB."));
                    }

                    subscriber.onNext(row);
                    subscriber.onCompleted();
                }
            }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}

你可以这样使用它：

storeToDb(storeThis)
        .subscribe(new Observer<Long>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                Log.e("STORING", "Something went south: " + e.getMessage());
            }

            @Override
            public void onNext(Long row) {
                Log.d("STORING", "Everything has been stored as record number: " + row);
            }
        });

是否需要退订

Is it necessary to unsubscribe

android

rx-java

rx-android