如何解决如何创建一个缓存每个计算项的 Observable? 等价于 Lazy<T>
我想创建一个能够缓存项目的序列 (Observable<T>
),因此管道内的计算只处理一次。例如:
var obs = Observable
.Range(1,100)
.SelectMany(x => GetItemAsync(x));
我希望有 2 个订阅者,GetItemAsync 的结果被缓存,因此第二个订阅者将从缓存的值中获取它们,因此对于任何后续订阅根本不应调用该方法。
我想做一些类似于 Lazy<T>
所做的事情,但是使用 Reactive Extensions
解决方法
Replay
运算符返回一个 IConnectableObservable<T>
,它是一个带有额外 IObservable<T>
方法的 Connect
。这个 observable 可以被任意数量的观察者订阅。它向所有观察者传播来自底层 observable 的所有过去和未来的通知,从连接时间开始到断开连接时间结束。下面是一个例子:
var connectable = Observable
.Range(1,100)
.SelectMany(x => GetItemAsync(x))
.Replay();
var subscription1 = connectable.Subscribe(x => Console.WriteLine(x))
var subscription2 = connectable.Subscribe(x => Console.WriteLine(x))
var connection = connectable.Connect(); // Subscribe to the 'SelectMany' observable
//...
connection.Dispose() // Unsubscribe from the 'SelectMany' observable
此示例演示了如何手动连接到底层 observable,这在使用其他多播运算符(如 Publish
)时很重要。但是对于 Replay
操作符,它不太重要,因为它具有重播功能:在它连接到底层 observable 之前还是之后订阅它并不重要。因此,您可以选择避免手动连接,而使用两个可用的自动连接运算符之一:
-
RefCount
:在第一次订阅时连接到底层 observable,并在最后一个订阅者取消订阅时断开连接。 -
AutoConnect(0)
:立即连接到底层 observable,并永远保持连接。
示例:
var observable = Observable
.Range(1,100)
.SelectMany(x => GetItemAsync(x))
.Replay()
.AutoConnect(0);
// call observable.Subscribe any time and as many times you want
当底层 observable 成功完成或出现错误时,RefCount
和 AutoConnect
也会自动断开连接。
不支持多次连接和断开连接,可能会产生意外结果。如果您想断开连接并重新连接,最好每次使用不同的 Replay
连接。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。