创建一个 IObservable 并立即返回缓存的异步操作的结果答案

【问题标题】：Create an IObservable and return the result of a cached async operation immediately创建一个 IObservable 并立即返回缓存的异步操作的结果
【发布时间】：2010-12-08 10:39:15
【问题描述】：

我正在使用响应式扩展来调用异步方法，我想缓存结果并将其返回以供后续调用该方法。

如何创建 Observable 实例，返回它并提供订阅所需的数据（cacheResult）？

public IObservable<Bar> GetBars(int pageIndex, int pageSize)
{
   var @params = new object[] { pageIndex, pageSize };
   var cachedResult = _cache.Get(@params);
   if (cachedResult != null)
   {
 // How do I create a Observable instance and return the 'cacheResult'...
 return ...
   }

   var observable = new BaseObservable<Bar>();
   _components.WithSsoToken(_configuration.SsoToken)
      .Get(@params)
      .Select(Map)
      .Subscribe(c =>
                     {
                          _cache.Add(@params, c);
                          observable.Publish(c);
                          observable.Completed();
                     }, exception =>
                     {
                        observable.Failed(exception);
                        observable.Completed();
                     });

       return observable;
}

【问题讨论】：

标签： c# .net system.reactive

【解决方案1】：

相信你在找Observable.Return:

return Observable.Return((Bar)cachedResult);

在不相关的注释上：

无需返回BaseObservable<T>。您应该返回一个 Subject<T>，因为它执行您的实现正在执行的操作，但它是线程安全的（您还应该在返回值上调用 .AsObservable()，因为它不能被强制回退）。
您使用Do 将值添加到缓存中：

var observable = new Subject<Bar>();
_components.WithSsoToken(_configuration.SsoToken)
    .Get(@params)
    .Select(Map)
    .Subscribe(c =>
    {
        _cache.Add(@params, c);
        observable.OnNext(c);
        observable.OnCompleted();
    }, exception =>
    {
        observable.OnError(exception);
    });

return observable.AsObservable();

【讨论】：

【解决方案2】：

为了方便，我已经为你编写了一个实现这种模式的类，请查看：

https://github.com/xpaulbettsx/ReactiveXaml/blob/master/ReactiveXaml/ObservableAsyncMRUCache.cs

var cache = new ObservableAsyncMRUCache<int, int>(
    x => Observable.Return(x*10).Delay(1000) /* Return an IObservable here */, 
    100 /*items to cache*/,
    5 /* max in-flight, important for web calls */
    );

IObservable<int> futureResult = cache.AsyncGet(10);

futureResult.Subscribe(Console.WriteLine);
>>> 100

它正确处理的一些棘手的事情：

它缓存最后 n 个项目并丢弃未使用的项目
它确保不超过 n 个项目同时运行 - 如果您不这样做，如果缓存为空，您很容易产生数千个 Web 调用
如果您连续两次请求同一个项目，第一个请求将发起一个请求，第二个调用将等待第一个请求，而不是产生相同的请求，因此您最终不会重复查询数据.

【讨论】：