如何在 Angular RxJS 中替换订阅中的 subscrib

可观测者是事件流.

远程调用(例如，调用服务器以发送邮件或调用数据库以清理某些数据)被实现为事件流，该事件流仅通知一个事件(即，远程调用的响应)，然后通知complete或仅通知错误错误(○)S.

使用RxJs操作符，您可以组合这样的流.例如，您可以执行以下操作:

• 您从一个流Stream_A开始，该流发出event_A，然后结束
• 您可以有第二个流Stream_B，它发出event_B，然后完成
• 然后组合Stream_A和Stream_B以创建第三个流Stream_A_B，第一个流triggers表示Stream_A的执行并发出event_A，一旦event_A被通知Stream_B的执行并发出Stream_B通知的所有事件，则Stream_B发出所有事件，在本例中为event_B
• 为了在RxJ中创建这个组合流，我们使用运算器concatMap(注意:人们通常使用switchMap来串联流--通常结果是相同的，但含义和潜在行为略有不同--调用远程服务的序列必须一个接一个地发生，concatMap通常是首选方法)

合并更多流以获得新流的另一个示例如下:

• 有两个流，Stream_1、Stream_2和Stream_3.这些流中的每个流发出一个值，然后完成.
• 我们可以将等待所有3个流发出和完成的这3个流组合在一起，然后只发出一个值，即流发出的所有值的数组，然后完成.
• 对于RxJ，利用函数forkJoin获得这种新的组合流

哈文说，希望对RxJ和可观测到的情况有所澄清，以下是我在您的情况下会做的事情

newYear(){
    // assume getAmicalistesValides returns an Observable that emits the result
    // of a remote call
    this.amicalisteService.getAmicalistesValides("")
    // to combine Observables we need to "pipe" operators, i.e. to execute
    // operators one after the other
    .pipe(
      // first thing to do here seems to send an email for each amicaliste
      // assuming we want to send the emails all in parallel, we can first
      // create one Observable for each mail to be sent and then use forkJoin
      // to execute them all in parallel
      // But all this has to happen after the Observable returned by getAmicalistesValides
      // has emitted its value, hence we use concatMap
      concatMap(amicalistes => {
        for(const amicaliste of amicalistes) {
          const to = amicaliste.email;
          const cc = null;
          const subject = "Adhère à l'AEIR";
          const body = ""
          // here we create the array of Observables
          const sendMailObs = this.amicalisteService.sendMail(null, to, cc, subject, body)
          // each of these Observables can print something or react to errors
          .pipe(tap({
            next: response => {
              console.log('E-mail envoyé avec succès !', response);
            },
            error: error => {
              console.error('Erreur lors de l\'envoi de l\'e-mail :', error);
            }))
          });
          // now we trigger the concurrent execution of all sendMail observables
          return forkJoin(sendMailObs)
      }),
      // after having sent the mails you want to do more stuff: delete data, images
      // and so on - assume each of these operations is an Observable
      // you will have to use concatMap and within it create the new Observables
      // and trigger them in parallel using forkJoin, as above
      concatMap(mailSentResults => {
         const deleteDataObs = ....
         const deleteImagesObs = ...
         ...
         return forkJoin([deleteDataObs, deleteImagesObs, // maybe other Obsevables])
      })
    )
    // up to here you have created a new stream, composing various other streams
    // and now is the time to subscribe to this new stream, which is the only stream 
    // you want to explicitely subscribe
    .subscribe({
      next: res => ... // manage the value notified by upstream
      error: err => ... // manage error
      complete: () => ... // do something when all is completed, if required
    })
  }

我希望我已经理解了你的情况，这一切都有一定的道理