原生Dart Streams中RxDart BehaviorSubject的替代方案：订阅获历史值

要在原生Dart Streams中实现RxDart里BehaviorSubject的核心功能——新订阅者能立即获取流中最近的一个历史值，可以通过自定义包装类封装StreamController来实现，以下是两种简洁的实现方式：

方法一：自定义BehaviorStream类 #

这个类会维护流的最新值，并在新订阅者加入时主动推送该值，之后再转发后续的新值：

class BehaviorStream<T> {
  final StreamController<T> _controller;
  T? _lastValue;

  BehaviorStream() : _controller = StreamController<T>.broadcast() {
    // 监听内部流，实时更新最新值
    _controller.stream.listen((value) => _lastValue = value);
  }

  // 向流中添加新值
  void add(T value) {
    _lastValue = value;
    _controller.add(value);
  }

  // 获取对外暴露的流，处理新订阅逻辑
  Stream<T> get stream {
    return Stream.multi((controller) {
      // 订阅时先推送历史最新值（如果存在）
      if (_lastValue != null) {
        controller.add(_lastValue!);
      }
      // 转发后续的新值
      final subscription = _controller.stream.listen(controller.add);
      // 订阅取消时清理资源
      controller.onCancel = () => subscription.cancel();
    });
  }

  // 关闭流控制器
  void close() => _controller.close();
}

使用示例 #

void main() async {
  final behaviorStream = BehaviorStream<int>();
  
  // 先添加几个值
  behaviorStream.add(1);
  behaviorStream.add(2);
  behaviorStream.add(3);
  
  // 此时订阅流
  behaviorStream.stream.listen((value) {
    print(value); // 先输出3，之后依次输出4、5
  });
  
  // 继续添加新值
  behaviorStream.add(4);
  behaviorStream.add(5);
  
  await Future.delayed(const Duration(seconds: 1));
  behaviorStream.close();
}

方法二：继承StreamController实现BehaviorController #

如果更倾向于直接扩展StreamController的行为，可以继承它并修改订阅逻辑：

class BehaviorController<T> extends StreamController<T> {
  T? _lastValue;

  BehaviorController() : super.broadcast() {
    // 监听自身流，保存最新值
    stream.listen((value) => _lastValue = value);
  }

  @override
  void add(T value) {
    _lastValue = value;
    super.add(value);
  }

  @override
  Stream<T> get stream {
    return Stream.multi((controller) {
      // 新订阅时推送历史值
      if (_lastValue != null) {
        controller.add(_lastValue!);
      }
      // 转发后续新值
      final sub = super.stream.listen(controller.add);
      controller.onCancel = sub.cancel;
    });
  }
}

使用方式 #

和普通StreamController几乎一致：

void main() async {
  final controller = BehaviorController<int>();
  
  controller.add(1);
  controller.add(2);
  controller.add(3);
  
  controller.stream.listen((value) {
    print(value); // 输出3、4、5
  });
  
  controller.add(4);
  controller.add(5);
  
  await Future.delayed(const Duration(seconds: 1));
  controller.close();
}

关键说明 #

• 以上实现均基于多订阅流（broadcast），和BehaviorSubject默认行为一致；如果需要单订阅版本，只需将StreamController.broadcast()改为普通的StreamController()即可。
• 核心逻辑是维护一个_lastValue变量保存最新值，在新订阅者通过Stream.multi创建流时，先推送该值，再转发后续的新值，完美匹配你需要的订阅时机获取历史最后值的需求。

内容的提问来源于stack exchange，提问作者Rza İsmayıl