RxJS BehaviorSubject状态更新顺序异常，如何实现顺序执行？

问题原因 #

BehaviorSubject 的 next() 是同步执行的。当你在某个订阅者的回调里调用 next() 时，会立刻中断当前的订阅通知流程，优先处理新值的订阅通知，等新通知走完后，才会回到之前没完成的旧通知流程。这就是为什么你会看到观察者接收值的顺序混乱——比如简化代码里 obs2 先收到 {a:2,b:3}，再收到 {a:1,b:1}，就是因为 obs1 处理旧值时同步触发了新的 next，打断了原有流程。

解决方案 #

方案1：用调度器延迟状态更新 #

把 next() 的调用放到异步队列里，让当前的订阅通知流程先完成，再处理新的状态更新。推荐用 queueScheduler（微任务队列，保持顺序且延迟极小）：

简化代码修改

import { BehaviorSubject, queueScheduler } from 'rxjs';

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

const obs1 = this.state$
  .subscribe((value) => {
    console.log('obs 1', value);
    if (value.a == 1) {
      // 把next放到微任务队列，等当前所有订阅者处理完旧值后再执行
      queueScheduler.schedule(() => {
        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 2 {a: 1, b: 1}
obs 1 {a: 2, b: 3}
obs 2 {a: 2, b: 3}

原始Angular代码修改

在触发 setState 的地方用调度器包裹：

import { queueScheduler } from 'rxjs';

// ... 其他组件代码不变

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

修改后日志输出：

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

方案2：用响应式操作符管理状态流（更推荐） #

从RxJS的响应式思维出发，避免在订阅回调里直接调用 next，而是把状态更新逻辑转化为流的一部分，用 concatMap 确保所有更新按顺序执行：

import { BehaviorSubject, concatMap, of } from 'rxjs';

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

// 将状态更新逻辑整合到流中，concatMap保证顺序执行
this.state$
  .pipe(
    concatMap(currentState => {
      console.log('processing state:', currentState);
      if (currentState.a === 1) {
        // 返回新的状态，concatMap会等待当前流处理完再发射新状态
        return of({ a: 2, b: 3 });
      }
      return of(currentState);
    })
  )
  .subscribe(updatedState => this.state$.next(updatedState));

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

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

这种方式完全遵循响应式编程范式，避免了同步副作用带来的顺序问题，代码也更易维护。

核心注意事项 #

• 绝对不要在同步订阅回调里直接调用 next，这会打破Observable的通知顺序规则。
• 用调度器或响应式操作符解决问题的本质，是让状态更新操作脱离当前同步执行栈，确保原有订阅通知流程完整执行后再处理新更新。

内容的提问来源于stack exchange，提问作者Franck