rxjs - 为什么 concatMap 没有按预期工作？

Question

请在下面查看我的代码

import { Observable, interval } from 'rxjs';
import { map, take, mergeMap, concatMap, switchMap, exhaustMap } from 'rxjs/operators';

const frameworks = ['Backbone', 'Angular', 'React', 'Vue'];

const getRecruits = agency => new Observable(observer => {
  console.log('agency', agency);
  interval(1000).pipe(
    take(5)
  ).subscribe(val => observer.next(`${agency} Developer ${val}`));
});

// concatMap
interval(3000).pipe(
  take(4),
  map(val => frameworks[val]),
  concatMap(agency => getRecruits(agency))
).subscribe(val => console.log(val));

这是我的输出：

我预计 Backbone 完成后，它会继续使用 Angular、React 和 Vue。但是在 Backbone 之后执行停止了

有什么解释吗？

Answer 1

要完成它，您需要调用observer.complete，目前代码observer.next仅调用。

要做到这一点，请像这样更改您的代码

).subscribe(
  val => observer.next(`${agency} Developer ${val}`),
  () => observer.complete(),
  () => observer.complete(),
);

Answer 2

关键知识

conactMap仅在前一个完成时发出下一个 observable。

原因

如果您牢记这一点并查看getRecruits，您会看到内部区间 observable 关闭，但实际上您新创建的new Observable(observer => ...)并没有关闭，因为您只向observervia发出值next。您可以在内部订阅触发完成的同时完成返回的 observable。

解决方案

const getRecruits = agency => new Observable(observer => {
  interval(1000).pipe(
    take(5)
  ).subscribe(
    val => observer.next(`${agency} Developer ${val}`),
    () => void 0, // You can use void 0 for no emit
    () => observer.complete() // Your interval source completes here and you call the new Observalbe to be completed
  );
});