有必要退订吗?

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

根据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线程是安全的。

在我看来，你应该总是有一个返回值。您的Store data to db路由肯定有一些返回值，例如指定行号的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);
            }
        });

Answer 2

当Observable完成后，RxJava会自动退订。您需要调用subscriber.onComplete()自动退订