angularjs - 缓存和延迟数据检索的可观察模式

Question

我正在尝试使用 RxJS Observable 在 Angular 中创建一个缓存函数。最初，我使用 angularjs $q 的延迟 promise 创建了这个方法。 Observables 和 RxJS 对我来说是新的，我发现这种工作方法仍然有些令人困惑。

这是我当前实现的 getOrCreate 缓存函数。从存储中检索键的单个值 (this.get())，如果它不在其中，则在其他地方检索它 (fetcher)。

假设 fetcher 是比 this.get() 慢的数据源。当我们仍在从 this.get() 检索时，可能会触发对同一 key 的多个请求，因此我放入了一个聚合器:为同一 key 的多个请求仅创建一个可观察值。

protected observableCache : {[key: string] : Observable<any>} = {};

get<T>(key : string): Observable<T> { /* Async data retrieval */ }

getOrCreate<T>(key : string, fetcher: () => Observable<T>) : Observable<T> {
  const keyHash = this.hash(key);

  // Check if an observable for the same key is already in flight
  if (this.observableCache[keyHash]) {
    return this.observableCache[keyHash];
  } else {

    let observable : Observable<T>;

    this.get(key).subscribe(

      // Cache hit
      (result) => { observable = Observable.of(result); },

      // Cache miss. Retrieving from fetching while creating entry
      () => {
        fetcher().subscribe((fetchedResult) => {
          if(fetchedResult) {
            this.put(key, fetchedResult);
          }
          observable = Observable.of(fetchedResult);
        });
      }
    );


   // Register and unregister in-flight observables
   this.observableCache[keyHash] = observable;
   observable.subscribe(()=> {
      delete this.observableCache[this.hash(key)];
   });

    return observable;
  }
}

这是我当前版本的代码，但看起来我没有正确处理异步代码:

• Observable 将在实例化之前返回:return observable 在 observable = Observable.of(result); 之前触发；
• 可能有一种更好的模式，可以在 this.get() 仍在运行时聚合对同一 key 的所有请求。

有人可以帮助找到应该使用的观察者模式吗？

Answer 1

我认为这可能有效。重写为:

getOrCreate<T>(key : string, fetcher: () => Observable<T>) : Observable<T> {
    const keyHash = this.hash(key);

    // Check if an observable for the same key is already in flight
    if (this.observableCache[keyHash]) {
        return this.observableCache[keyHash];
    }

    let observable : ConnectableObservable<T> = this.get(key)
        .catch(() => { // Catch is for when the source observable throws  error: It replaces it with the new Rx.Observable that is returned by it's method
            // Cache miss. Retrieving from fetching while creating entry
            return this.fetchFromFetcher(key, fetcher);
        })
        .publish();

    // Register and unregister in-flight observables
    this.observableCache[keyHash] = observable;
    observable.subscribe(()=> {
        delete this.observableCache[keyHash];
    });
    observable.connect();

    return observable;
},

fetchFromFetcher(key : string, fetcher: () => Observable<T>) : Observable<T> {
    // Here we create a stream that subscribes to fetcher to use `this.put(...)`, returning the original value when done
    return Rx.Observable.create(observer => {
        fetcher().subscribe(fetchedResult => {
            this.put(key, fetchedResult);
            observer.next(fetchedResult);
        },
        err => observer.error(err),
        () => observer.complete())
    });
}

说明:

1. 可观察量与 promise 有很大不同。他们要处理异步的东西，有一些相似之处，但又截然不同
2. 由于 this.get(...) 看起来是异步的，您的 let observable 在产生一个值之前不会被填充，因此当您将其分配给您的缓存为空是正常的。
3. 可观察量的一个伟大之处(以及与 promise 的主要区别)是您可以在执行任何内容之前定义一个流。在我的解决方案中，在调用 observable.connect() 之前不会调用任何内容。这避免了太多的.subscriptions
4. 因此，在我的代码中，我获取了 this.get(key) 流，并告诉它如果失败 (.catch(...))必须获取结果，但一旦获取，请将其放入缓存中 (this.put(key, fetchedResult)
5. 然后我publish()这个可观察的:这使得它的行为更像 promise ，它使它变得“热门”。这意味着所有订阅者都将从同一个流中获取值，而不是每次订阅时都创建一个从 0 开始的新流。
6. 然后我将其存储在可观察池中，并设置在完成后将其删除。
7. 最后，我.connect()。仅当您 publish() 时才会完成此操作，它是实际订阅原始流并执行您想要的所有内容的东西。

为了明确起见，因为这是来自 Promises 的常见错误，如果您将流定义为:

let myRequest = this.http.get("http://www.example.com/")
    .map((result) => result.json());

请求尚未发送。每次执行 myRequest.subscribe() 时，都会向服务器发出新请求，它不会重用“第一次订阅”结果。这就是 .publish() 非常有用的原因:它使得当您调用 .connect() 时，它会创建一个触发请求的订阅，并将共享最后收到的结果(Observables 支持流:许多结果)以及对已发布 Observable 的所有传入订阅。