如何创建一个缓存每个计算项的 Observable？（等价于 Lazy<T>）

Question

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

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

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

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

Answer 1

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 运算符的重放功能，它不太重要：在它连接到底层可观察对象之前或之后是否被订阅并不重要。因此，您可以选择避免手动连接，并使用两个可用的自动连接运算符之一：

RefCount: 第一次订阅时连接到底层可观察对象，最后一个订阅者取消订阅时断开连接。
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

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

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

如何创建一个缓存每个计算项的 Observable？（等价于 Lazy<T>）

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

.net

c#

system.reactive

rx.net

如何创建一个缓存每个计算项的 Observable？ （等价于 Lazy<T>）

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

.net

c#

system.reactive

rx.net

如何创建一个缓存每个计算项的 Observable？（等价于 Lazy<T>）