如何使用 RxJs 在 Angular 中限制并发 API 请求

Question

我有一个 Angular 7 应用程序，它允许您多次上传文件。对于选择的每个文件，它都会向服务器发出请求。这不是很好，因为打开数百个并发调用对我的后端服务器来说似乎是一个潜在的问题。

我想做的是限制我的应用程序可以发出的并发请求数。我有一个通用 API 类，我想用它来限制应用程序范围内的限制，而不仅仅是文件上传，而不是需要文件上传组件本身来管理它。

诚然，我有时对 RxJx 感到困惑，但我很确定这是可能的。

class ApiService {

    get(path: string, params: any = {}): Observable<any> {
        return this.http.get(path`, { params: params });
    }
    
    uploadFile(path: string, body: any = {}): Observable<any> {
        ...code for preparing file here...
        return this.http.post(path, body);
    }
    
}

class FileUploader {

    // called many times-- once for each file
    uploadFile(file) {
        this.apiService.uploadFile(path, body: file).subscribe(response => {
             // use response here
        })
    }
}

我想象的是，在 api 类中，我可以添加到使用最大并发或其他东西的队列中，而不是立即在 fileUpload 或 get 函数中执行 http 调用，并等待调用直到有空间.但我不确定如何执行此操作，因为我已立即订阅 File Uploader 类。

Answer 1

您可以使用Subject 和mergeMap 运算符

interface FileUpload {
   path: string;
   body: file;
}

export class UploadService {
  private readonly CONCURRENT_UPLOADS = 2;
  private uploadQ = new Subject<FileUpload>();

  constructor(private api: ApiService) {
    this.uploadQ.asObservable().pipe(
      mergeMap(fu => this.api.uploadFile(fu.path, fu.body)).pipe(
        // must catch error here otherwise the subscriber will fail
        // and will stop serving the Q
        catchError(err => {
          console.error('Caught error ', err);
          return of(err);
        })), this.CONCURRENT_UPLOADS),
    ).subscribe((res: WhateverResultYouGet) => {
         // process result  
      }, err => {
        // something went wrong
      });
  }

  // this is your original signature of the method but where do you get path, actually?
  /**
  * Push the file to upload Q
  */
  uploadFile(file) {
    this.uploadQ.next({path, body: file});
  }
}

您只需将上传推送到队列中，而不是立即启动上传。队列由构造函数中的订阅提供服务，使用 mergeMap 运算符，您可以在其中实际指定并发性。