angular - 等待可观察量完成后再调用下一个

Question

我按照以下方式使用 RxJS，并且得到了意外的值顺序。以下代码可以在Stackblitz中执行。

编辑 - 简化的代码。我们可以简化为同一行为主体的 2 个简单观察者，它们以不同的顺序接收数据。

this.state$ = new BehaviorSubject<any>({ a: null, b: null });

const obs1 = this.state$
  .subscribe((value) => {
    console.log('obs 1', value);
    if (value.a == 1) this.state$.next({ a: 2, b: 3 });
  });

const obs2 = this.state$.subscribe((value) => console.log('obs 2', value));

this.state$.next({ a: 1, b: 1 });

这就是日志的样子

obs 1 {a: null, b: null}
obs 2 {a: null, b: null}
obs 1 {a: 1, b: 1}
obs 1 {a: 2, b: 3}
obs 2 {a: 2, b: 3}
obs 2 {a: 1, b: 1}

我希望相同行为主体的观察者能够以相同的顺序接收发出的值，无论这些值来自何处。

之前的代码

class sampleState {
  a: any;
  b: any;
}

const initialState: sampleState = {
  a: null,
  b: null,
};

@Component({
  selector: 'my-app',
  standalone: true,
  imports: [CommonModule],
  template: ``,
})
export class App {
  name = 'Angular';

  protected _state$: BehaviorSubject<sampleState>;
  state$: Observable<sampleState>;

  constructor() {
    this._state$ = new BehaviorSubject<sampleState>(initialState);
    this.state$ = this._state$.asObservable();

    this.select((state) => state.a).subscribe((value) => {
      console.log('a ' + value);
      if (value == 1) this.setState({ b: 3 });
    });

    this.select((state) => state.b).subscribe((value) => {
      console.log('b ' + value);
    });

    this._state$.subscribe((value) => console.log(value));

    this.setState({ a: 1, b: 1 });
  }

  public get state() {
    return this._state$.getValue();
  }

  public setState(newState: any) {
    this._state$.next({
      ...this.state,
      ...newState,
    });
  }

  public select<K>(mapFn: (state: sampleState) => K): Observable<K> {
    return this._state$.asObservable().pipe(
        map((state: sampleState) => mapFn(state)),
       distinctUntilChanged()
    );
  }
}

我收到以下日志

{a: null, b: null}
{a: 1, b: 3}
{a: 1, b: 1}

虽然我期望

{a: null, b: null} // initialisation
{a: 1, b: 1} // setState
{a: 1, b: 3} // if a == 1 => set b = 3

在调用下一个 setState 之前，第一个 setState 似乎尚未完成。有没有办法改变这种行为，使状态按顺序改变？

Answer 1

您看到此行为是因为所有操作都是同步的，并且您从“选择一个”订阅中发出一个新值，这会导致订阅逻辑递归执行。

要防止递归行为，您可以使用 asapScheduler像这样:

this.state$ = this._state$.pipe(observeOn(asapScheduler));

现在，更新您的 select 方法以使用 this.state$ 而不是 this._state$ ，您应该可以开始了。

这是一个StackBlitz你可以去看看。