如何创建一个缓存每个计算项的 Observable? (等价于 Lazy<T>)

How to create an Observable that caches each of the calculated items? (equivalence of Lazy<T>)

我想创建一个能够缓存项目的序列 (Observable<T>),因此管道内的计算只处理一次。例如:

var obs = Observable
.Range(1, 100)
.SelectMany(x => GetItemAsync(x));

我希望有 2 个订阅者,缓存 GetItemAsync 的结果,这样第二个订阅者将从缓存的值中获取它们,因此根本不应该为任何后续订阅调用该方法。

我想做一些类似于 Lazy<T> 所做的事情,但使用 Reactive Extensions

Replay 运算符 returns 一个 IConnectableObservable<T>,这是一个带有额外 Connect 方法的 IObservable<T>。这个可观察对象可以被任意数量的观察者订阅。它向所有观察者传播来自底层可观察对象的所有过去和未来的通知,从它连接的时间开始,到它断开连接的时间结束。这是一个例子:

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

此示例演示了如何手动连接到底层可观察对象,这在使用 Publish 等其他多播运算符时很重要。但由于 Replay 运算符的重放功能,它不太重要:在它连接到底层可观察对象之前或之后是否被订阅并不重要。因此,您可以选择避免手动连接,并使用两个可用的自动连接运算符之一:

  1. RefCount: 第一次订阅时连接到底层可观察对象,最后一个订阅者取消订阅时断开连接。

  2. AutoConnect(0):立即连接到底层可观察对象,并永远保持连接。

示例:

var observable = Observable
    .Range(1, 100)
    .SelectMany(x => GetItemAsync(x))
    .Replay()
    .AutoConnect(0);

// call observable.Subscribe any time and as many times you want

当底层可观察对象成功完成或出现错误时,RefCountAutoConnect 也会自动断开连接。

不支持多次连接和断开连接,并且可能会产生意外结果。如果您想断开连接并重新连接,最好每次都使用不同的 Replay 连接器。