Angular / RxJS - 在应用程序初始化期间不要重复相同的 GET 请求

Question

Angular 或 RxJS 中是否有任何运算符，如果它们在短时间内被调用，可以将多个 http 请求合二为一？

---

F.e.当我打开我的应用程序时，ngOnInit 期间的所有 20 个组件都在进行相同的 http 调用：

我已经基于 Angular HttpInterceptor (f.e. https://blog.logrocket.com/caching-with-httpinterceptor-in-angular/) 创建了一个缓存，但是因为所有这些调用都是在初始化时执行的，所以缓存的构建速度不够快。

我可以将所有组件放在一个父级中，这将通过输入将数据传递给子级，但我想保留架构，组件可以在其中初始化自身（它们在其他地方被重用）。

我尝试使用 RxJS 共享或共享回放，但我想在 f.e. 之后停止共享。 1秒。我很乐意让这样的运营商为我服务：

getData(): Observable<any> {
    return = this.http.get<any>('endpoint')
      .pipe(
        shareIfCalledWithin( 1000ms ),
      );

Answer 1

父组件是提供共享数据是个好主意。

另一种通用解决方案是将http调用委托给subject并从subject触发http调用，subject触发频率可以由debounceTime控制

const getData = new Subject()
getData.next()
    
const getDataStream = getData.pipe(debounceTime(1000),
switchMap(_ => this.http.get......),
share())
    
getDataStream.subscribe(data => you data...)

您可以将此逻辑放在父组件或共享服务中。

Answer 2

我倾向于认为你在服务中推送的逻辑越多越好，这种情况似乎是这种方法的一个很好的候选者。

更具体地说，正如@Fan Cheung 已经建议的那样，我将创建一个服务（通过依赖注入共享），它公开 2 个 API

• 一种触发远程服务执行的方法
• 一个 Observable，一旦服务调用到达，它就会发出结果

所以服务的代码可能是这样的

export class MyService {
  // define a private Subject and then export it as an Observable
  // this Subject emits any time a result is received from the remote service
  // you can use a ReplaySubject if you want chaching
  private _dataFromRemoteService$ = new Subject<any>()
  // API clients can subscribe to to receive the data
  public dataFromRemoteService$ = _dataFromRemoteService$.asObservable()

  // private Subject used as trigger for the fetch operation
  private _trigger$ = new Subject<any>();
  // API to be called to trigger the execution of the remote service
  public fetchData() {
    _trigger.next();
  }
  // private method that subscribes to the trigger and uses debounce to control the number of calls to remote service
  private _fetchData() {
     this._trigger.pipe(
        debounceTime(1000),
        switchMap(this.http.get<any>('endpoint')
     )
     // the result received from the remote service is broadcasted to all subscribers of dataFromRemoteService$
     .subscribe(_dataFromRemoteService$)
  }
  
  constructor() {
     // start the subscription
     _fetchData();
  }
}

现在任何 Component 订阅 dataFromRemoteService$ 以接收准备好的数据，并可以调用方法 fetchData 来触发远程服务的执行。

export class MyComponent implements OnInit {
  constructor(private service: MyService) {}

  ngOnInit(): void {
     // Subscribe to the stream of results of the execution of the service
     this.service.dataFromRemoteService$.subscribe(
        data => {// do something with the data received from remote service}
     )

     // trigger the execution of the service
     this.service.fetchData();
  }
}

这种方法的主要优点之一是您可以在不同的情况下轻松测试服务，例如许多客户端以快速的顺序请求远程数据。

Answer 3

谢谢，与此同时，我想出了对缓​​存服务的升级，但由于它似乎过于工程化，我可能会接受其他答案。不过，我相信下面的代码将来可能会对某人有所帮助。

一般情况下，具有特定类型（urlWithParams）的第一个请求存储在pending中。在此请求完成之前，相同类型的进一步请求将存储在waiting 中，并返回相应的新的、空的主题和来自该主题的可观察对象。当挂起的请求完成时，它会向等待的此类主体发送正确的 HttpResponse。

@Injectable()
export class HttpCacheInterceptor implements HttpInterceptor {
  constructor(private cache: CoreHttpCacheService) {}

  intercept(req: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
    if( req.method !== 'GET' && req.method !== 'OPTIONS' ) {
      this.cache.clear();
    }

    if( req.method !== 'GET' || req.headers.get( CC.NO_CACHE ) ) {
      return next.handle(req);
    }

    const cachedResponse: HttpResponse<any> = this.cache.get(req.urlWithParams);

    if( this.cache.pending.has( req.urlWithParams ) ) {
      const sbj = new Subject<HttpEvent<any>>();
      this.cache.waiting.push( {r: req.urlWithParams, s: sbj} );
      return sbj.asObservable();
    } else {
      if( cachedResponse ) {
        return of( cachedResponse.clone() );
      } else {
        if( ! this.cache.pending.has( req.urlWithParams ) ) {
          this.cache.pending.add( req.urlWithParams );
        }
        return next.handle(req).pipe(
          tap(stateEvent => {
            if(stateEvent instanceof HttpResponse) {
              this.cache.set(req.urlWithParams, stateEvent.clone());

              let i = this.cache.waiting.length;
              while( i-- ) {
                if( this.cache.waiting[i].r === req.urlWithParams ) {
                  this.cache.waiting[ i ].s.next( stateEvent.clone() );
                  this.cache.waiting.splice( i, 1 );
                }
              }

              this.cache.pending.delete( req.urlWithParams );
            }
          })
        );
      }
    }
  }
}


@Injectable({
  providedIn: 'root'
})
export class CoreHttpCacheService {
  pending = new Set< string >();
  waiting: {r: string, s: Subject<HttpEvent<any>> }[] = [];

  private cache: Map< string, HttpResponse<any> > = new Map();

  private timers = new Map< string, NodeJS.Timer >();

  get( key: string ) {
    return this.cache.get( key );
  }

  set( key: string, v: any ) {
    this.timers.set( key, setTimeout( () => this.clear( key ), Constants.TIME_100SEC ) );
    return this.cache.set( key, v );
  }

  clear( key?: string ) {
    if( key !== undefined ) {
      this.cache.delete( key );
    } else {
      this.cache.clear();
    }
  }
}

此示例将因 HTTP 错误而失败，仅通过 urlWithParams 比较请求，并且 CoreHttpCacheService 字段是公开的，因此如果您想使用它，请小心。