如何用 rxjs 一个接一个地进行 2 个异步调用，而不管第一个是成功还是失败

Question

我有一个用例，我需要进行两次异步调用。顺序在这里很重要。在开始下一个呼叫之前，第一个呼叫需要完成（或失败）。

我在下面有一个示例代码显示了我需要什么，但它不是以反应方式完成的（即使用 rxjs 运算符）

public performDBTasks(): Observable<void> {
    return Observable.create(
    (observer)=> {
        this.performDBTask1().subscribe(
        ()=> {
            this.performDBTask2().subscribe(
            ()=> {
                observer.next();
                observer.complete();
            },
            ()=> {
                observer.error();
            });
        },
        (error)=> {
            this.performDBTask2().subscribe(
            ()=> {
                observer.next();
                observer.complete();
            },
            ()=> {
                observer.error();
            });         
        }); 
    });
}

Answer 1

您可以使用 switchMap 和 last 来解决您的问题。

const { Subject, merge, of} = rxjs;
const { switchMap, last, map, catchError } = rxjs.operators;

const first$ = new Subject();
const second$ = new Subject();

const result$ = first$.pipe(
  // catch an error and return an observable that gets "?" and completes afterwards
  catchError(() => of("?")),
  // only emit if observable completes and return last value
  last(),
  // switch to observable second$
  switchMap((firstValue) =>
    second$.pipe(
      // just for logging purpose - log the first$ and second$ value
      map((secondValue) => [firstValue, secondValue])
    )
  )
);

result$.subscribe(console.log);

first$.next(1);
first$.next(2);
first$.complete();
second$.next(4);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>

信息

1. 根据你的问题，我不知道你想要达到什么确切的输出。您的问题仅描述了异步调用顺序。如果输出不符合您的需求，请发表评论，我可以尝试进行调整。但请记住指定/更新您的问题。
2. 您的示例代码订阅了许多未处理的订阅。最好避免尽可能多的订阅。完美的是，您最多可以为每个 ui 更新或后端/服务调用订阅一次。在这个问题中阅读更多关于how to avoid memory leaks with rxjs pipes
3. *无法通过

   first$.error("msg")

   创建错误。相应的 stackoverflow 问题：如何在行为主题上抛出错误并继续流。随意模拟管道上的错误。不需要模拟来回答问题，所以我将避免实施模拟。

Answer 2

您可以将

from

与

concatMap

一起使用，它会一直等到上一个请求完成。

const { from, concatMap, fetch: { fromFetch }, tap } = rxjs;

const endpoints = [
  // Returns 404
  'https://dummy.restapiexample.com/api/v1/404-error',
  // Returns 200 
  'https://dummy.restapiexample.com/api/v1/employee/1'
];

from(endpoints).pipe(
  // Won't fetch until the previous request has finished.
  concatMap(url => fromFetch(url).pipe(
    tap(url => console.log(`Making request to: ${url.url}`)),
  )),
  tap(resp => console.log(resp.status))
).subscribe()

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.8.0/rxjs.umd.min.js"></script>