如果 rxjs 仍在进行数据加载，如何返回数据或等待答案

【问题标题】：How to return data or wait if data loading is still in progress with rxjs如果 rxjs 仍在进行数据加载，如何返回数据或等待
【发布时间】：2020-04-22 21:49:36
【问题描述】：

我有一个服务，它使用 Observable 在它的构造函数中加载一些数据。然后在稍后的某个时间，可以使用 getter 检索数据。如果数据存在，它应该立即返回数据。或者，如果仍在进行中，请等待加载完成。我想出了以下示例（代码在 Typescript 中）：

class MyService {
    private loadedEvent = new Subject<any>();
    private loaded = false;
    private data: any;

    constructor(
        private dataService: DataService
    ) {
        this.dataService.loadData().subscribe(data => {
            this.data = data;
            this.loaded = true;
            this.loadedEvent.next(null);
        });
    }

    public getDataOrWait(): Observable<any> {
        if (this.loaded) { // A
            return of(this.data);
        }
        return new Observable((observer: Observer<any>) => {
            const subscription = this.loadedEvent.subscribe(() => { // B
                observer.next(this.data);
                observer.complete();
                subscription.unsubscribe();
            });
        });
    }
}

有没有更简单的方法来做到这一点？这一定是一种常见的模式。

另外，我认为如果在标记为 A 和 B 的行之间的某处执行加载完成，则存在竞争条件（我不确定此处是否涉及线程 - 但是数据是异步加载的）。

【问题讨论】：

标签： angular typescript rxjs

【解决方案1】：

您所要做的就是使用shareReplay() 运算符：

class MyService {
    public data$: Observable<any>;
    public loaded$: Observable<boolean>;

    constructor(private dataService: DataService) {
        this.data$ = this.dataService.loadData().pipe(
            shareReplay(1);
        );
        this.loaded$ = this.data$.pipe(
           mapTo(true),
           startWith(false)
        );
    }
}

shareReplay 运算符是一个多播运算符，它将向所有订阅者发出相同的先前值。订阅者将等到第一个值可用。

然后您可以使用该data$ 创建一个loaded$ observable，它将发出false，直到data$ 最终发出一个值，然后当值准备好时，它将始终发出true。

或者，您可以让data$ 在数据准备好之前发出null，然后是数据。下游有一些合乎逻辑的好处，允许您创建新的 observables 以了解数据何时准备就绪。

        this.data$ = this.dataService.loadData().pipe(
            startWith(null),
            shareReplay(1);
        );

您必须调用myService.data$.subscribe() 来触发流的第一次读取以使数据准备好。您可以在构造函数中执行此操作，但请记住，Angular 在首次使用之前不会创建服务。如果您希望快速加载数据，请在路由中使用解析器或将服务注入NgModule 构造函数并在那里订阅。

【讨论】：

精彩的答案。谢谢！我都试过了，你的和@Myk Willis 的答案。我很难决定为接受的答案选择哪个。对于可能略有不同的要求，我喜欢这两种方法。最后我选了他，只是因为他是第一个。
@nharrer 感谢您的投票。我更关心选票而不是接受。只要我帮助别人了解运营商，我就很高兴。

【解决方案2】：

您似乎只是想在逻辑上将数据服务的基于 Observable 的接口扩展到 MyService 类的客户端。您可以使用新的AsyncSubject，它会在完成后向所有订阅者发出一个值。

class MyService {
  private data: any;
  private dataSubject = new AsyncSubject<any>();

  constructor(
    private dataService: DataService
  ) {
    this.dataService.loadData().subscribe(data => {
      this.data = data;
      this.dataSubject.next(data);
      this.dataSubject.complete();
    });
  }

  public getData(): Observable<any> {
    return this.dataSubject.asObservable();
  }
}

getData 的调用者随后会执行以下操作：

    service.getData().subscribe((data) => {
      console.log(`got data ${data}`);
    });

【讨论】：

非常感谢。我不知道 AsyncSubject。这很好用。只有一个小问题。如果 this.data 稍后更改怎么办（我的示例中没有这种情况）。由于 dataSubject 已经完成，所以值永远不会改变。
@nharrer 如果您希望 getData() 例程始终返回最新数据（而不是假设数据只设置一次），请查看 BehaviorSubject。
是的，我试过了。但由于 BehaviorSubject 始终具有初始值，因此无需等待加载完成。但这导致我使用 ReplaySubject(1)，它的行为类似于 AsyncSubject 但之后可以调用 .next()。

【解决方案3】：

首先，您应该知道 observable 保证来自服务器的任何响应对您来说都是存在的，这样您的设计就会考虑到这一点，只需在订阅中处理您的逻辑，然后从该订阅中转移。这一切都是在异步模式下完成的。

但是，如果您只获得单个结果而不是流式传输，并且想要等到数据完全从服务器加载，那么您可以使用带有 sync await 的 Promise（这是常见的模式）来同步得到结果。顺便说一句，您的场景听起来像是 Promise 行为。

如果你必须使用 Observables，你可以使用 forkJoin 或 flatMap 等待日期完全加载。

通过完美比较检查这个很棒的答案

Is it a good practice using Observable with async/await?

【讨论】：