angular - 使用 RxJS(和 Angular)临时缓存来自参数化请求的 HTTP 响应

Question

我想缓存 HTTP 响应并使之过期带有特定参数的 GET 请求。

用例示例:

假设我构建这样的服务:

@Injectable()
export class ProductService {

  constructor(private http: HttpClient) {}

  getProduct$(id: string): Observable<Product> {
    return this.http.get<Product>(`${API_URL}/${id}`);
  }

  getProductA$(id: string): Observable<ProductA> {
    return this.getProduct$(id).pipe(
      // bunch of complicated operations here
    };
  }

  getProductB$(id: string): Observable<ProductB> {
    return this.getProduct$(id).pipe(
      // bunch of other complicated operations here
    };
  }

}

现在无论出于何种原因，函数 A 在组件 A 中调用，函数 B 在组件 B 中调用。我知道这可以通过另一种方式完成(例如顶级智能组件获取 HTTP 数据并将其传递给输入参数)，但无论出于何种原因，这两个组件都是“智能”的，它们各自调用一个服务函数。

两个组件都加载到同一个页面上，所以两个订阅发生 = 对同一个端点的两个 HTTP 请求 - 即使我们知道结果很可能是相同的。

我想简单地缓存 getProduct$ 的响应，但我也希望此缓存很快过期，因为 2 分钟后，产品管理部门的 Margareth 将更改产品价格。

我尝试过但不起作用的方法:

基本上，我尝试使用窗口时间为 5 秒的 shareReplay 保留热可观察对象的字典。我的假设是，如果(源)可观察对象完成或订阅数为 0，那么下一个订阅者将简单地重新触发可观察对象，但情况似乎并非如此。

private product$: { [productId: string]: Observable<Product> } = {};

getProduct$(id: string): Observable<Product> {
  if (this.product$[id]) {
    return this.product$[id];
  }
  this.product$[id] = this.http.get<Product>(`${API_URL}/${id}`)
    .pipe(
      shareReplay(1, 5000), // expire after 5s
    )
  );
  return this.product$[id];
}

我想，我可以尝试在完成时使用 finalize 或 finally 从我的字典中删除 observable，但不幸的是，每次取消订阅时也会调用这些。

所以解决方案可能更复杂。

有什么建议吗？

Answer 1

2022 年 7 月编辑:由于原始解决方案仅适用于较旧的 RxJS 版本，并且基本上基于 RxJS 中的错误，因此 RxJS 7.0+ 具有相同的功能:

import { of, defer, share, delay, tap, timestamp, map, Observable } from 'rxjs';

let counter = 1;

const mockHttpRequest = () =>
  defer(() => {
    console.log('Making request...');
    return of(`Response ${counter++}`).pipe(delay(100));
  });

const createCachedSource = (
  makeRequest: () => Observable<any>,
  windowTime: number
) => {
  let cache;

  return new Observable((obs) => {
    const isFresh = cache?.timestamp + windowTime > new Date().getTime();
    // console.log(isFresh, cache);

    if (isFresh) {
      obs.next(cache.value);
      obs.complete();
    } else {
      return makeRequest()
        .pipe(
          timestamp(),
          tap((current) => (cache = current)),
          map(({ value }) => value)
        )
        .subscribe(obs);
    }
  }).pipe(share());
};

const cached$ = createCachedSource(() => mockHttpRequest(), 1000);

// Triggers the 1st request.
cached$.subscribe(console.log);
cached$.subscribe(console.log);
setTimeout(() => cached$.subscribe(console.log), 50); 
setTimeout(() => cached$.subscribe(console.log), 200);

// Triggers the 2nd request.
setTimeout(() => cached$.subscribe(console.log), 1500);
setTimeout(() => cached$.subscribe(console.log), 1900);
setTimeout(() => cached$.subscribe(console.log), 2400);

// Triggers the 3nd request.
setTimeout(() => cached$.subscribe(console.log), 3000);

现场演示:https://stackblitz.com/edit/rxjs-rudgkn?devtoolsheight=60&file=index.ts

原始答案:如果我对您的理解正确，您希望根据响应的 id 参数缓存响应，所以当我制作两个 getProduct() 时使用不同的 id 我会得到两个不同的未缓存结果。

我认为最后一个变体几乎是正确的。您希望它取消订阅其父项，以便稍后可以重新订阅并刷新缓存值。

如果我没记错的话，shareReplay 运算符在 RxJS 5.5 之前的工作方式有所不同，其中 shareReplay 已更改并且它没有重新订阅其源代码。它后来在 RxJS 6.4 中重新实现，您可以根据传递给 shareReplay 的配置对象修改其行为。由于您使用的是 shareReplay(1, 5000) 看起来您使用的是 RxJS <6.4 所以最好使用 publishReplay() 和 refCount() 运算符代替。

private cache: Observable<Product>[] = {}

getProduct$(id: string): Observable<Product> {
  if (!this.cache[id]) {
    this.cache[id] = this.http.get<Product>(`${API_URL}/${id}`).pipe(
      publishReplay(1, 5000),
      refCount(),
      take(1),
    );
  }

  return this.cache[id];
}

请注意，我还包含了 take(1)。那是因为我希望链在 publishReplay 发出其缓冲区之后和它订阅其源 Observable 之前立即完成。没有必要订阅它的来源，因为我们只想使用缓存的值。 5 秒后缓存值被丢弃，publishReplay 将再次订阅它的源。

我希望这一切都有意义 :)。