如何创建一个缓存每个计算项的 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 之前还是之后订阅它并不重要。因此，您可以选择避免手动连接，而使用两个可用的自动连接运算符之一：

1. RefCount：在第一次订阅时连接到底层 observable，并在最后一个订阅者取消订阅时断开连接。

2. 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 连接。