如何在 RxJava 中刷新 observable？

Question

我有一个 observable，它封装了一个 HTTP 请求

mObservable =  retryObservable(mService.getAddressList(getUserId(), true, 1, Integer.MAX_VALUE, "id", true)
            .map(r -> {
                return r.getItems();
            })
            .observeOn(AndroidSchedulers.mainThread()));

然后订阅

mSubscription = mObservable.subscribe(items -> {
                mAddressAdapter.swapItems(items);
            }, getActivityBase()::showError);

当订阅初始化到来时，cold observable 被激活并触发 HTTP 请求。现在，我知道基础数据已更改，我需要提出相同但新的请求。 我试过了

mSubscription.unsubscribe();

然后调用

mObservable.subscribe(items -> {doSomething();})

再次，据我了解，订阅应该会触发 observable，但它不起作用。 有什么建议吗？

Answer 1

Observable 完成后不会发布任何新项目。这是 Rx 合约。

将您的代码包装到一个方法中，并且每次都创建一个新的 observable。

Observable<?> getObservable() {
    return retryObservable(mService.getAddressList(getUserId(), true, 1, Integer.MAX_VALUE, "id", true)
            .map(r -> {
                return r.getItems();
            })
            .observeOn(AndroidSchedulers.mainThread()));
}

正如@DaveSexton 在评论中提到的那样，在 RxJava 中使用 defer 函数有更好的方法

在订阅者订阅之前不要创建 Observable；创建一个 每次订阅的新 Observable

传递 defer() 一个 Observable 工厂函数（一个函数 生成 Observables)，而 defer() 将返回一个 Observable 将调用此函数重新生成其 Observable 序列 每次有新订阅者订阅。

更多：https://github.com/ReactiveX/RxJava/wiki/Creating-Observables#defer