在异步调用中创建的 RxJava Observable 替代方案答案

【问题标题】：RxJava Observable alternative to create in async call在异步调用中创建的 RxJava Observable 替代方案
【发布时间】：2016-08-14 09:40:21
【问题描述】：

我听了这个演讲 https://www.youtube.com/watch?v=QdmkXL7XikQ&feature=youtu.be&t=274

听说我应该避免使用 create 方法创建 Observable，因为它不会自动处理取消订阅和背压，但我在下面的代码中找不到替代方法。

compositeSubscription.add(
    Observable.create(new Observable.OnSubscribe<DTOCompaniesCallback>() {
        @Override
        public void call(final Subscriber<? super DTOCompaniesCallback> subscriber) {

            modelTrainStrike.getCompaniesFromServer(new CompaniesCallback() {
                @Override
                public void onResult(DTOCompaniesCallback dtoCompaniesCallback) {
                    try {
                        if (!subscriber.isUnsubscribed()) {
                            subscriber.onNext(dtoCompaniesCallback);
                            subscriber.onCompleted();
                        }
                    } catch (Exception e) {
                        if (!subscriber.isUnsubscribed()) {
                            subscriber.onError(e);
                        }
                    }
                }
            });

        }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<DTOCompaniesCallback>() {
        @Override
        public void call(DTOCompaniesCallback dtoCompaniesCallback) {
            Log.i("TAG", "onResult: " + dtoCompaniesCallback.getCompaniesList().size());
        }
    }, new Action1<Throwable>() {
        @Override
        public void call(Throwable throwable) {
            throw new OnErrorNotImplementedException("Source!", throwable);
        }
    })
);

我在 OnDestroy 方法中调用 clear CompositeSubscription

@Override
public void onDestroy() {
    if (compositeSubscription != null) {
        compositeSubscription.clear();
    }
}

您是否看到我可以在此处使用的 create 方法的替代方法？您是否看到任何潜在危险或这种方法是否安全？谢谢

【问题讨论】：

标签： java rx-java observable frp

【解决方案1】：

你可以使用 defer + AsyncSubject:

Observable.defer(() -> {
    AsyncSubject<DTOCompaniesCallback> async = AsyncSubject.create();
    modelTrainStrike.getCompaniesFromServer(v -> {
        async.onNext(v);
        async.onComplete();
    });
    return async;
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
...

如果getCompaniesFromServer 支持取消，您可以：

Observable.defer(() -> {
    AsyncSubject<DTOCompaniesCallback> async = AsyncSubject.create();
    Closeable c = modelTrainStrike.getCompaniesFromServer(v -> {
        async.onNext(v);
        async.onComplete();
    });
    return async.doOnUnsubscribe(() -> {
        try { c.close(); } catch (IOException ex) { }
    });
})

【讨论】：