rxdart BehaviorSubject输出顺序不符合预期问题咨询

输出顺序和预期不符，核心原因是Dart流的异步调度机制与BehaviorSubject的实现逻辑共同作用，和官方定义的「新订阅立即收到最新值」特性并不冲突。

核心调度规则 #

• BehaviorSubject默认是异步广播流：所有事件派发动作（包括add触发的多监听器回调、新订阅时给新监听器推送当前最新值的动作）都不会在代码调用处同步执行，而是会被放入Dart的微任务队列，等当前同步执行块全部跑完后，才会按照入队顺序依次执行。
• 测试代码从启动到调用subject.close()的全部逻辑都在同一个同步块里，这期间没有任何print回调被执行，所有待触发的回调都在排队：
  1. 初始化种子值为0的Subject后，注册listener1，「给listener1发0」的回调入队
  2. 调用add(1)，「给已存在的listener1发1」的回调入队
  3. 调用add(2)，「给已存在的listener1发2」的回调入队
  4. 注册listener2，「给listener2发当前最新值2」的回调入队
  5. 调用add(3)，「给已存在的listener1、listener2分别发3」的两个回调依次入队
  6. 注册listener3，「给listener3发当前最新值3」的回调入队
  7. 调用add(4)，「给已存在的listener1、listener2、listener3分别发4」的三个回调依次入队
  8. 同步代码执行完毕，开始执行微任务队列，受Dart广播流的监听器遍历顺序、微任务调度优先级影响，最终就会出现看似乱序的输出。

符合预期的验证方式 #

如果要看到和文档描述一致的、顺序可预测的输出，只需要在每一步操作后等待微任务队列清空即可，测试代码如下：

import 'package:rxdart/rxdart.dart';

void main() async {
  BehaviorSubject<int> subject = BehaviorSubject<int>.seeded(0);

  subject.stream.listen((a) => print("listener 1 : $a"));
  await Future.delayed(Duration.zero); // 等待队列执行完成

  subject.add(1);
  await Future.delayed(Duration.zero);
  subject.add(2);
  await Future.delayed(Duration.zero);

  subject.stream.listen((a) => print("listener 2 : $a"));
  await Future.delayed(Duration.zero);

  subject.add(3);
  await Future.delayed(Duration.zero);

  subject.stream.listen((a) => print("listener 3 : $a"));
  await Future.delayed(Duration.zero);

  subject.add(4);
  await Future.delayed(Duration.zero);

  subject.close();
}

运行后输出完全符合官方定义的行为：

listener 1 : 0
listener 1 : 1
listener 1 : 2
listener 2 : 2
listener 1 : 3
listener 2 : 3
listener 3 : 3
listener 1 : 4
listener 2 : 4
listener 3 : 4

使用注意事项 #

• 新订阅立即收到当前最新值是BehaviorSubject的固有行为，和调度模式无关，不会因为异步调度失效。
• 如果业务场景需要无延迟的同步事件派发，可以在构造Subject时传入sync: true参数，此时add触发的事件会同步派发给所有已注册监听器，不需要等待微任务调度，但要注意同步流禁止在监听器回调中反向触发流的添加、取消订阅操作，否则会抛出递归异常。

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