掌握 Dart 的异步数据流——让数据"推"向 UI 而非 UI "拉"取数据

目录

  1. 为什么需要 Stream：从 E-Brufen 的"播放状态同步"说起
  2. Dart Stream 核心概念全景
  3. Stream 的基本操作：从 listen 到 transform
  4. StreamTransformer 与流处理管道
  5. 实战1：将 MoodStorage 改为 Stream 驱动
  6. 实战2：搜索防抖实现——Dart 原生方式与 RxDart 方式
  7. 实战3：多数据源组合——combineLatest 实战
  8. RxDart 增强：BehaviorSubject、ReplaySubject 与更多操作符
  9. Stream 常见错误与避坑指南
  10. Stream vs ChangeNotifier vs Future：三大响应式方案对比
  11. 总结

一、为什么需要 Stream：从 E-Brufen 的"播放状态同步"说起

在这里插入图片描述

1.1 一个真实的需求

在 E-Brufen 白噪音功能中，有一个看似简单实则复杂的需求：鸿蒙原生端（ArkTS）的音频播放器状态变化，需要实时同步到 Flutter（Dart）端的 UI。

具体场景如下：

用户在鸿蒙控制中心按下"暂停"
    │
    ▼
ArkTS AVPlayer.on('stateChange') 触发
    │
    ▼
需要通知 Dart 端：播放已暂停，更新 UI（停止脉冲动画、显示暂停按钮）

这是一个典型的"事件源不在一处，消费者在另一处"的场景。ChangeNotifier 能胜任吗？

如果使用 ChangeNotifier，我们可以在原生端收到事件后，通过 MethodChannel 调用 Dart 端的一个方法，方法内部调用 notifyListeners()。这当然可行。但问题是：原生端可能在任意时刻触发事件——用户按暂停、系统中断音频、蓝牙耳机断开等——这些事件天然是一个"流"而非"单次通知"。

用 ChangeNotifier 模拟数据流：notifyListeners() 每次触发，UI 需要主动去拉取当前状态（Pull 模式）。但原生端的状态变化是一个 Push 模式——“我变了，你更新”——这正好是 Stream 的核心语义。

1.2 E-Brufen 中的 Stream 实际使用

打开 lib/data/ohos_audio.dart，可以看到我们已经用 Stream 解决了这个问题：

class OhosAudioPlayer {
  static const _channel = MethodChannel('com.ebrufen/audio_player');

  // ── 核心：广播流，承载原生端回传的播放状态事件 ──
  static final Stream<PlaybackEvent> onPlaybackStateChanged =
      _playbackController.stream;
  static final _playbackController =
      StreamController<PlaybackEvent>.broadcast();

  static bool _handlerRegistered = false;

  static void _ensureHandler() {
    if (_handlerRegistered) return;
    _handlerRegistered = true;
    _channel.setMethodCallHandler((call) async {
      switch (call.method) {
        case 'onPlaybackStateChanged':
          final state = (call.arguments as Map?)?.cast<String, dynamic>();
          final stateStr = state?['state'] as String? ?? '';
          // ── 关键：将原生事件推入 Stream ──
          _playbackController.add(PlaybackEvent.fromString(stateStr));
          break;
      }
    });
  }
}

在 UI 层（SoundscapePage），我们订阅这个流：

_playbackSub = OhosAudioPlayer.onPlaybackStateChanged.listen((event) {
  if (!mounted) return;
  switch (event.state) {
    case PlaybackState.playing:
      setState(() { _isPlaying = true; _pulse.repeat(reverse: true); });
      break;
    case PlaybackState.paused:
    case PlaybackState.stopped:
      setState(() { _isPlaying = false; _pulse.stop(); _pulse.reset(); });
      break;
  }
});

整个链条：

原生 ArkTS AVPlayer 状态变化
  └─► MethodChannel.callHandler (Dart)
       └─► _playbackController.add(event)    // 推入 Stream
            └─► onPlaybackStateChanged.listen  // 所有订阅者收到事件
                 └─► setState() 更新 UI

这就是 Stream 在 E-Brufen 中发挥的真实作用——不只是一个语言特性，而是连接原生层和 Flutter UI 层的"事件动脉"。

二、Dart Stream 核心概念全景

2.1 四个核心角色

Dart 的 Stream 体系由四个角色组成：

┌─────────────────────────────────────────────────────────────────────┐
│                     Dart Stream 体系架构                              │
│                                                                      │
│  ┌──────────────────┐      ┌──────────────────┐                      │
│  │ StreamController │      │    Stream<T>     │                      │
│  │                  │      │                  │                      │
│  │ .add(event) ─────┼─────►│ 数据流管道        │                      │
│  │ .addError(err) ──┤      │ .listen()        │                      │
│  │ .close() ────────┤      │ .map() .where()  │                      │
│  │ .stream ─────────┘      │ .transform()     │                      │
│  └──────────────────┘      └────────┬─────────┘                      │
│                                     │                                │
│                          ┌──────────▼──────────┐                     │
│                          │ StreamSubscription │                     │
│                          │                    │                      │
│                          │ .pause()           │                      │
│                          │ .resume()          │                      │
│                          │ .cancel()          │                      │
│                          │ .onData()          │                      │
│                          │ .onError()         │                      │
│                          │ .onDone()          │                      │
│                          └──────────┬─────────┘                     │
│                                     │                                │
│                          ┌──────────▼──────────┐                     │
│                          │   UI / Business     │                     │
│                          │   处理数据的代码      │                     │
│                          └─────────────────────┘                    │
└─────────────────────────────────────────────────────────────────────┘

StreamController（流的控制器）：流的"入口"。你通过 add() 向流中推入数据，通过 close() 关闭流。

Stream（数据流）：数据的"传输管道"。它不存储数据，而是传递数据。你可以对流进行变换（map、where、transform），也可以订阅它（listen）。

StreamSubscription（订阅）：流的"出口"。通过 listen() 获得，用于接收数据、处理错误、以及取消订阅。

数据消费者：接收流数据的代码——UI setState、业务逻辑处理、状态更新等。

2.2 单订阅流 vs 广播流

这是 Dart Stream 设计中最重要的二分法：

// ── 单订阅流（Single-Subscription Stream） ──
// 只能有一个监听者。适合"一次性"的数据流：
// 文件读取、HTTP 响应、手势序列等。
final controller = StreamController<int>();  // 默认是单订阅
final stream = controller.stream;

stream.listen((data) => print('监听者 1: $data'));
// stream.listen((data) => print('监听者 2: $data'));  // ❌ 运行时错误！
// StateError: Stream has already been listened to.

// ── 广播流（Broadcast Stream） ──
// 可以有多个监听者。适合"事件总线"式场景：
// 播放状态变化、WebSocket 消息、用户登录/登出事件。
final broadcastController = StreamController<int>.broadcast();
final broadcastStream = broadcastController.stream;

broadcastStream.listen((data) => print('监听者 1: $data')); // ✅
broadcastStream.listen((data) => print('监听者 2: $data')); // ✅

选择决策表：

场景推荐流类型原因
文件读取结果单订阅只有一个消费者关心文件内容
HTTP 请求响应单订阅一次请求对应一个消费者
原生层 → Flutter 的事件广播多个页面可能需要监听同一事件
用户登录状态广播导航栏、个人页、设置页都可能关心
WebSocket 消息广播连接只有一个，但消息需要分发给多个处理器
数据库变更通知广播多个 UI 组件关注同一数据源的变更

E-Brufen 的 onPlaybackStateChanged 使用广播流，正是因为它需要被多个页面（白噪音页面、可能的锁屏控制页面）同时监听。

2.3 StreamController 的三个生命周期方法

final controller = StreamController<String>();

// 1. add() — 推入数据
controller.add('事件 1');
controller.add('事件 2');

// 2. addError() — 推入错误
controller.addError(Exception('播放失败'));

// 3. close() — 关闭流
controller.close();
// 关闭后不能再 add，所有订阅者收到 onDone 回调

StreamController 的内部实现使用了一个回调列表。当 add() 被调用时，Controller 遍历所有活跃的订阅，调用它们的 onData 回调。这与 ChangeNotifier 的 notifyListeners() 遍历 _listeners 极其相似。

2.4 两种创建 Stream 的方式

除了 StreamController，Dart 还提供了直接创建 Stream 的方式：

// 方式一：Stream.fromFuture —— 将 Future 转为单事件 Stream
final stream1 = Stream.fromFuture(fetchUserData());
// 发出 Future 的结果，然后自动关闭

// 方式二：Stream.periodic —— 周期性地发出事件
final ticker = Stream.periodic(const Duration(seconds: 1), (count) => count);
// 每 1 秒发出 count（0, 1, 2, 3, ...）

// 方式三：Stream.fromIterable —— 从集合创建 Stream
final stream3 = Stream.fromIterable([1, 2, 3, 4, 5]);
// 依次发出 1, 2, 3, 4, 5，然后关闭

// 方式四：async* 生成器 —— 用同步语法写异步流
Stream<int> countDown(int from) async* {
  for (int i = from; i >= 0; i--) {
    await Future.delayed(const Duration(seconds: 1));
    yield i;  // 每 1 秒发出一个值
  }
}

2.5 Stream 与 Future 的根本区别

维度FutureStream
值数量一个零个、一个或多个
完成语义成功或失败，然后结束持续发出事件，直到被关闭或取消
可否取消否（Future 一旦开始就无法取消）是（cancel() 取消订阅）
典型场景HTTP 请求、文件读取、数据库查询实时事件、定时器、用户输入、WebSocket
类比一个承诺（Promise）一个事件管道（Event Pipe）
内存模型结果到达后 Future 对象可被 GC只要 Stream 未关闭且有人订阅，持有引用

简单记忆：Future 是一次性的异步结果，Stream 是持续性的异步事件序列。

三、Stream 的基本操作：从 listen 到 transform

3.1 listen() —— 流的入口

final stream = Stream.fromIterable([1, 2, 3, 4, 5]);

final subscription = stream.listen(
  (data) {
    print('收到数据: $data');       // onData: 处理每个事件
  },
  onError: (error) {
    print('发生错误: $error');      // onError: 处理错误事件
  },
  onDone: () {
    print('流已关闭');              // onDone: 流正常关闭时调用
  },
  cancelOnError: false,            // true: 遇到第一个错误就取消订阅
);

3.2 流的变换操作

Stream 支持一系列函数式变换操作，这些操作返回新的 Stream，使得你可以构建数据处理管道：

final source = Stream.fromIterable([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

// map: 转换每个元素
source
    .map((n) => n * 2)               // 2, 4, 6, 8, 10, 12, ...
    .listen(print);

// where: 过滤元素
source
    .where((n) => n % 2 == 0)        // 2, 4, 6, 8, 10
    .listen(print);

// expand: 一对多展平
source
    .expand((n) => [n, n * 10])      // 1, 10, 2, 20, 3, 30, ...
    .listen(print);

// take: 只取前几个
source
    .take(3)                          // 1, 2, 3
    .listen(print);

// skip: 跳过前几个
source
    .skip(5)                          // 6, 7, 8, 9, 10
    .listen(print);

// distinct: 去重（连续重复）
Stream.fromIterable([1, 1, 2, 2, 2, 3, 1])
    .distinct()                       // 1, 2, 3, 1  (注意：最后的 1 保留，因为它不紧跟另一个 1)
    .listen(print);

// asyncMap: 异步映射（每个元素可以执行异步操作）
source
    .asyncMap((n) async {
      await Future.delayed(const Duration(milliseconds: 100));
      return '处理完成: $n';
    })
    .listen(print);

// handleError: 错误处理
source
    .map((n) => n == 5 ? throw Exception('不能是 5') : n)
    .handleError((error) => print('捕获错误: $error'))
    .listen(print);

3.3 在 E-Brufen 中应用流变换

假设我们要对播放状态事件流进行过滤和映射：

// 原始流：所有播放事件
OhosAudioPlayer.onPlaybackStateChanged

// 只关心 "播放中" 和 "已暂停" 状态
    .where((event) =>
        event.state == PlaybackState.playing ||
        event.state == PlaybackState.paused)

// 转换为 UI 可用的布尔值
    .map((event) => event.state == PlaybackState.playing)

// 去重：连续相同的状态不需要重复通知
    .distinct()

// 订阅
    .listen((isPlaying) {
      setState(() => _isPlaying = isPlaying);
    });

仅用 4 个链式调用，我们就将原生端传来的"播放状态变化事件"转换成了 UI 需要的"是否正在播放"布尔值。这就是 Stream 管道的威力。

3.4 流的暂停与恢复

StreamSubscription 提供了精细的流量控制：

final sub = stream.listen(handler);

sub.pause();               // 暂停接收事件（事件在内部缓冲）
sub.resume();              // 恢复接收（缓冲的事件会立即交付）
sub.pause(Future.delayed(...));  // 暂停，等待一个 Future 完成后再恢复
sub.cancel();              // 永久取消订阅

在 E-Brufen 中，当 BreathePage 暂停呼吸练习时，理论上可以使用 pause() 来暂停 Timer 产生的事件流。但我们实际上使用了更简单的方式——在 Timer 回调中检查 _isPaused 标志位来跳过处理。两种方式各有优劣：

方式优点缺点
subscription.pause()事件不会触发回调，彻底静默暂停期间的事件会缓冲在内存中
标志位（_isPaused）简单直观，不缓冲事件回调依然被调用（浪费 CPU）
subscription.cancel() + 重建彻底释放资源重建有开销

对于高频事件（如 Timer 每秒触发），推荐标志位方式。对于低频事件（如用户点击），推荐 pause/resume。

四、StreamTransformer 与流处理管道

4.1 为什么需要 StreamTransformer

map、where 等操作符覆盖了 80% 的变换需求，但有时你需要更复杂的变换逻辑——比如：

  • 将多个事件聚合成一个事件（buffer）
  • 根据当前事件和历史事件做决策（stateful transformation）
  • 在变换中引入计时逻辑（debounce、throttle）

这时就需要 StreamTransformer。

4.2 StreamTransformer 的基本结构

/// 自定义 StreamTransformer：将数值流每 3 个一组打包
class BufferTransformer<T> implements StreamTransformer<T, List<T>> {
  final int bufferSize;

  BufferTransformer(this.bufferSize);

  @override
  Stream<List<T>> bind(Stream<T> stream) {
    final buffer = <T>[];
    return stream.transform(StreamTransformer<T, List<T>>.fromHandlers(
      handleData: (data, sink) {
        buffer.add(data);
        if (buffer.length >= bufferSize) {
          sink.add(List<T>.from(buffer));
          buffer.clear();
        }
      },
      handleDone: (sink) {
        if (buffer.isNotEmpty) {
          sink.add(List<T>.from(buffer));
        }
        sink.close();
      },
    ));
  }
}

// 使用
Stream.fromIterable([1, 2, 3, 4, 5, 6, 7, 8])
    .transform(BufferTransformer<int>(3))
    .listen(print);
// 输出: [1, 2, 3]
//       [4, 5, 6]
//       [7, 8]

4.3 实战：DebounceTransformer —— 防抖变换器

防抖是搜索功能的标配——用户快速输入时，我们不想每次按键都触发搜索，而是等待用户停止输入 300ms 后再执行：

/// 防抖变换器：在静默期之后才发出最近的一个事件。
///
/// 如果新事件在静默期内到来，旧的计时器被取消，重新开始计时。
class DebounceTransformer<T> implements StreamTransformer<T, T> {
  final Duration duration;

  DebounceTransformer(this.duration);

  @override
  Stream<T> bind(Stream<T> stream) {
    Timer? timer;
    return stream.transform(StreamTransformer<T, T>.fromHandlers(
      handleData: (data, sink) {
        timer?.cancel();       // 取消旧计时器
        timer = Timer(duration, () {
          sink.add(data);      // 静默期过后，发出事件
        });
      },
      handleDone: (sink) {
        timer?.cancel();
        sink.close();
      },
    ));
  }
}

// 使用——完美适配搜索输入
_searchController.textChanges
    .transform(DebounceTransformer<String>(
      const Duration(milliseconds: 300),
    ))
    .listen((query) {
      performSearch(query);
    });

4.4 实战：ThrottleTransformer —— 节流变换器

与防抖不同，节流保证在固定时间间隔内最多发出一个事件：

/// 节流变换器：在指定时间间隔内，只发出第一个事件，忽略后续事件。
class ThrottleTransformer<T> implements StreamTransformer<T, T> {
  final Duration duration;
  DateTime? _lastEmit;

  ThrottleTransformer(this.duration);

  @override
  Stream<T> bind(Stream<T> stream) {
    return stream.transform(StreamTransformer<T, T>.fromHandlers(
      handleData: (data, sink) {
        final now = DateTime.now();
        if (_lastEmit == null ||
            now.difference(_lastEmit!) >= duration) {
          _lastEmit = now;
          sink.add(data);
        }
      },
    ));
  }
}

4.5 流处理管道模式

多个 Transformers 可以串联成管道，形成数据处理流水线：

原始事件流
  │
  ▼
  where: 过滤无关事件
  │
  ▼
  DebounceTransformer: 防抖（等待静默期）
  │
  ▼
  ThrottleTransformer: 节流（限制频率）
  │
  ▼
  map: 转换数据格式
  │
  ▼
  asyncMap: 异步处理（如调用 API）
  │
  ▼
  listen: 最终消费

/// 搜索输入处理管道
Stream<String> buildSearchPipeline(Stream<String> rawInput) {
  return rawInput
      .transform(DebounceTransformer<String>(
        const Duration(milliseconds: 300),
      ))
      .where((query) => query.length >= 2)    // 至少 2 个字符才搜索
      .distinct()                                // 相同查询不重复
      .asyncMap((query) => searchAPI(query));   // 调用搜索 API
}

五、实战1：将 MoodStorage 改为 Stream 驱动

5.1 当前 ChangeNotifier 方案的局限

E-Brufen 当前的 MoodStorage 使用 ChangeNotifier 模式：

class MoodStorage extends ChangeNotifier {
  Future<int> insert(MoodEntry entry) async {
    // ... 写入逻辑 ...
    notifyListeners();    // 广播：数据变了
    return id;
  }

  Future<void> delete(int id) async {
    // ... 删除逻辑 ...
    notifyListeners();    // 广播：数据变了
  }
}

UI 层通过 addListener / removeListener 模式订阅：

widget.moodStorage.addListener(_loadMoods);

void _loadMoods() {
  setState(() {
    _allMoods = widget.moodStorage.getAll();   // 主动拉取（Pull）
    _weekMoods = widget.moodStorage.getByWeek(DateTime.now());
  });
}

这种方案的问题：

  1. Pull 模式：UI 需要主动调用 getAll() 拉取全部数据，即使只有一条记录被修改
  2. 无变更详情：notifyListeners() 不携带任何信息——UI 不知道是什么操作、哪条记录、发生了什么变化
  3. 无法区分操作类型：插入、更新、删除都触发同一个回调

5.2 基于 Stream 的重构方案

定义 Mood 变更事件：

/// 情绪数据的变更事件
enum MoodChangeType { inserted, updated, deleted }

class MoodEvent {
  final MoodChangeType type;
  final int moodId;
  final MoodEntry? entry;  // inserted/updated 时附带数据

  const MoodEvent({
    required this.type,
    required this.moodId,
    this.entry,
  });
}

改造 MoodStorage，添加 Stream 出口：

class MoodStorage extends ChangeNotifier {
  static const _boxName = 'moods';
  Box? _box;
  int _nextId = 1;

  // ── 新增：Stream 控制器 ──
  final _eventController = StreamController<MoodEvent>.broadcast();

  /// 数据变更事件流——替代 ChangeNotifier 的细粒度方案
  Stream<MoodEvent> get events => _eventController.stream;

  // ── CRUD（改造后） ──

  Future<int> insert(MoodEntry entry) async {
    final id = _nextId++;
    final data = entry.toJson();
    data['id'] = id;
    await _box?.put(id, jsonEncode(data));

    // 同时发出 ChangeNotifier 通知（向后兼容）和 Stream 事件
    notifyListeners();
    _eventController.add(MoodEvent(
      type: MoodChangeType.inserted,
      moodId: id,
      entry: entry,
    ));
    return id;
  }

  Future<void> update(int id, MoodEntry updated) async {
    final data = updated.toJson();
    data['id'] = id;
    await _box?.put(id, jsonEncode(data));

    notifyListeners();
    _eventController.add(MoodEvent(
      type: MoodChangeType.updated,
      moodId: id,
      entry: updated,
    ));
  }

  Future<void> delete(int id) async {
    await _box?.delete(id);

    notifyListeners();
    _eventController.add(MoodEvent(
      type: MoodChangeType.deleted,
      moodId: id,
    ));
  }

  // ── 原有查询方法不变 ──
  List<MoodEntry> getAll() { /* ... */ }
  List<MoodEntry> getByDate(DateTime date) { /* ... */ }

  @override
  void dispose() {
    _eventController.close();   // ← 关键：关闭 StreamController
    _box?.close();
    super.dispose();
  }
}

5.3 UI 层按事件类型做增量更新

有了 MoodEvent，UI 层不再需要全量刷新：

class _DiaryPageState extends State<DiaryPage> {
  StreamSubscription<MoodEvent>? _moodSub;

  @override
  void initState() {
    super.initState();
    _tabController = TabController(length: 3, vsync: this);

    // ── 订阅 Mood 变更事件流 ──
    _moodSub = widget.moodStorage.events.listen((event) {
      if (!mounted) return;

      switch (event.type) {
        case MoodChangeType.inserted:
          // 增量：在列表头部插入新记录
          setState(() {
            _allMoods.insert(0, event.entry!);
          });
          break;

        case MoodChangeType.updated:
          // 增量：更新列表中对应的记录
          setState(() {
            final index = _allMoods.indexWhere((m) => m.id == event.moodId);
            if (index != -1 && event.entry != null) {
              _allMoods[index] = event.entry!;
            }
          });
          break;

        case MoodChangeType.deleted:
          // 增量：从列表中移除
          setState(() {
            _allMoods.removeWhere((m) => m.id == event.moodId);
          });
          break;
      }

      // 周统计可能变了，刷新之
      _weekMoods = widget.moodStorage.getByWeek(DateTime.now());
    });

    // 首次全量加载
    _allMoods = widget.moodStorage.getAll();
    _weekMoods = widget.moodStorage.getByWeek(DateTime.now());
  }

  @override
  void dispose() {
    _moodSub?.cancel();    // ← 关键：取消订阅
    _tabController.dispose();
    _noteController.dispose();
    super.dispose();
  }
}

5.4 重构前后对比

维度ChangeNotifier（重构前）Stream（重构后）
数据传递方式Pull：UI 主动拉取Push：事件携带数据
操作区分无（notifyListeners 不传参数）有（inserted/updated/deleted）
刷新粒度全量刷新（getAll 遍历 Hive）增量更新（O(1) 列表操作）
100 条记录的一次 insert全量遍历 100 条 + 排序直接在列表头部插入 1 条
订阅管理addListener / removeListenerlisten / cancel
内存安全容易忘记 removeListenerStreamBuilder 自动管理
调试可见性难以追踪谁触发了通知可以在 Stream 上打日志

5.5 渐进式迁移策略

E-Brufen 不需要一次性将所有 ChangeNotifier 替换为 Stream。一个实际的迁移策略是：

// 阶段一：双轨运行（当前方案 + Stream）
class MoodStorage extends ChangeNotifier {
  final _eventController = StreamController<MoodEvent>.broadcast();
  // 同时保留 ChangeNotifier 和 Stream，零风险
}

// 阶段二：新功能优先使用 Stream
// 新增的"搜索心情记录"功能直接用 Stream

// 阶段三：旧代码逐步迁移
// 将 DiaryPage 从 addListener 迁移到 Stream 订阅

六、实战2：搜索防抖实现——Dart 原生方式与 RxDart 方式

6.1 E-Brufen 的搜索需求

假设我们要为 E-Brufen 的日记时间线添加搜索功能——用户输入关键词，实时过滤匹配的心情记录。原始实现可能是这样的：

// ❌ 每次输入都触发搜索——性能浪费
TextField(
  onChanged: (query) {
    setState(() {
      _filteredMoods = _allMoods
          .where((m) => m.note?.contains(query) == true)
          .toList();
    });
  },
);

问题：用户快速输入"今天心情不错"，触发了 7 次过滤操作（每输入一个字一次）。虽然 E-Brufen 的本地数据集很小（<500 条），但在真实搜索中（如搜索网络资源或大型数据集），这是一个严重的性能问题。

6.2 Dart 原生防抖实现

class _DiarySearchState extends State<DiarySearch> {
  final _searchController = TextEditingController();
  Timer? _debounceTimer;

  @override
  void initState() {
    super.initState();
    _searchController.addListener(_onSearchChanged);
  }

  void _onSearchChanged() {
    // 防抖核心：每次新输入取消旧计时器，重新开始计时
    _debounceTimer?.cancel();
    _debounceTimer = Timer(const Duration(milliseconds: 300), () {
      if (!mounted) return;
      final query = _searchController.text;
      _performSearch(query);
    });
  }

  void _performSearch(String query) {
    setState(() {
      _filteredMoods = query.isEmpty
          ? _allMoods
          : _allMoods
              .where((m) =>
                  m.note?.toLowerCase().contains(query.toLowerCase()) == true)
              .toList();
    });
  }

  @override
  void dispose() {
    _debounceTimer?.cancel();         // ← 别忘取消 Timer
    _searchController.removeListener(_onSearchChanged);
    _searchController.dispose();
    super.dispose();
  }
}

6.3 使用 Stream 实现防抖

将 TextEditingController 的文本变化转换为 Stream：

// TextEditingController 本身没有 Stream 暴露文本变化
// 但我们可以自己创建一个
class _DiarySearchState extends State<DiarySearch> {
  final _searchController = TextEditingController();
  final _queryController = StreamController<String>.broadcast();
  StreamSubscription? _searchSub;

  @override
  void initState() {
    super.initState();
    _searchController.addListener(() {
      _queryController.add(_searchController.text);
    });

    // 构建防抖搜索管道
    _searchSub = _queryController.stream
        .transform(DebounceTransformer<String>(
          const Duration(milliseconds: 300),
        ))
        .where((q) => q.length >= 2 || q.isEmpty)  // 空或>=2字符才搜索
        .distinct()                                  // 相同不重复
        .listen((query) {
          _performSearch(query);
        });
  }

  @override
  void dispose() {
    _searchSub?.cancel();
    _queryController.close();
    _searchController.dispose();
    super.dispose();
  }
}

6.4 使用 RxDart 实现防抖

如果你愿意引入 rxdart 包，代码会更加简洁：

import 'package:rxdart/rxdart.dart';

class _DiarySearchState extends State<DiarySearch> {
  final _searchController = TextEditingController();
  final _querySubject = BehaviorSubject<String>();  // RxDart 的 Subject
  StreamSubscription? _searchSub;

  @override
  void initState() {
    super.initState();
    _searchController.addListener(() {
      _querySubject.add(_searchController.text);
    });

    _searchSub = _querySubject
        .debounceTime(const Duration(milliseconds: 300))  // ← RxDart 内置防抖
        .where((q) => q.length >= 2 || q.isEmpty)
        .distinct()
        .listen((query) {
          _performSearch(query);
        });
  }

  @override
  void dispose() {
    _searchSub?.cancel();
    _querySubject.close();
    _searchController.dispose();
    super.dispose();
  }
}

6.5 三种防抖方案对比

方案代码量外部依赖可测试性灵活性推荐场景
Timer 手动实现中（~20行）0中低单个防抖需求
Dart 原生 Stream + Transformer高（~40行）0高高需要管道的场景，零依赖要求
RxDart低（~10行）rxdart高最高复杂响应式逻辑，团队已有 RxDart 经验

对于 E-Brufen 这个只需单个防抖的应用，Timer 方案是最务实的选择。但如果你在构建一个需要大量流操作（组合、合并、防抖、节流、缓冲）的应用，RxDart 值得引入。

七、实战3：多数据源组合——combineLatest 实战

7.1 问题场景

在 E-Brufen 中，日记统计页面需要同时依赖两个数据源：

  1. 情绪数据流（MoodStorage）：用户随时可能添加/删除记录
  2. 时间范围（用户选择的日期）：用户可能切换"本周"/“本月”

这两个数据源是独立的——任何一个变化都需要重新计算统计数据。传统方案是在两个数据源各自的回调中都调用同一个更新方法，但这会导致协调问题（如果两个数据源几乎同时变化，可能会触发两次重复计算）。

7.2 Dart 原生方案：手动协调

class _StatsPageState extends State<StatsPage> {
  List<MoodEntry> _moods = [];
  DateTime _selectedWeek = DateTime.now();
  Timer? _recomputeDebounce;

  @override
  void initState() {
    super.initState();
    // 监听情绪数据变化
    widget.moodStorage.addListener(_onMoodsChanged);
    _onMoodsChanged();
  }

  void _onMoodsChanged() {
    _moods = widget.moodStorage.getAll();
    _scheduleRecompute();
  }

  void _onWeekChanged(DateTime week) {
    _selectedWeek = week;
    _scheduleRecompute();
  }

  void _scheduleRecompute() {
    _recomputeDebounce?.cancel();
    _recomputeDebounce = Timer(const Duration(milliseconds: 50), () {
      _recompute();
    });
  }

  void _recompute() {
    // 基于 _moods 和 _selectedWeek 重新计算统计
    final weekMoods = _moods.where((m) => isInWeek(m.createdAt, _selectedWeek)).toList();
    setState(() {
      _weekStats = computeStats(weekMoods);
    });
  }
}

这个方案工作正常，但有明显的"手工"感——在用 Timer 手动做协调。这不是一个好的抽象。

7.3 RxDart 方案：combineLatest

RxDart 的 combineLatest 操作符完美解决了这个问题——它监听多个流，当其中任何一个流发出新值，就用所有流的最新值组合出一个新结果：

import 'package:rxdart/rxdart.dart';

class StatsViewModel {
  // 两个独立的 Subject
  final _moodsSubject = BehaviorSubject<List<MoodEntry>>.seeded([]);
  final _weekSubject = BehaviorSubject<DateTime>.seeded(DateTime.now());

  // 输入流
  void updateMoods(List<MoodEntry> moods) => _moodsSubject.add(moods);
  void updateWeek(DateTime week) => _weekSubject.add(week);

  // 组合输出流
  Stream<WeekStats> get weekStats => Rx.combineLatest2(
    _moodsSubject,
    _weekSubject,
    (List<MoodEntry> moods, DateTime week) {
      final weekMoods = moods.where(
        (m) => isInWeek(m.createdAt, week),
      ).toList();
      return computeStats(weekMoods);  // 只在任一输入变化时执行一次
    },
  );

  void dispose() {
    _moodsSubject.close();
    _weekSubject.close();
  }
}

在 UI 层使用：

class StatsPage extends StatefulWidget {
  final StatsViewModel viewModel;
  // ...
}

class _StatsPageState extends State<StatsPage> {
  StreamSubscription<WeekStats>? _statsSub;

  @override
  void initState() {
    super.initState();
    _statsSub = widget.viewModel.weekStats.listen((stats) {
      setState(() => _weekStats = stats);
    });
  }

  @override
  void dispose() {
    _statsSub?.cancel();
    super.dispose();
  }
}

7.4 combineLatest 的原理图示

_moodsSubject:   ──[M1]─────[M2]────────────────────[M3]──►
_weekSubject:    ──────[W1]──────────[W2]──►

组合结果:         ───────────────────────────────►
                  当两个流都有初始值后才开始发出组合结果

combineLatest:   ──────[R1]────[R2]───[R3]──[R4]──►
                        ↑        ↑       ↑      ↑
                   M1+W1    M2+W1   M2+W2  M3+W2

关键特性：

  • 只有当所有输入流都至少发出过一个值时，才开始产生组合结果
  • 任意一个流发出新值，就用所有流的最新值组合一次
  • 不会重复触发——两个流同时变化时（同一帧内），只会触发一次

7.5 更多 RxDart 组合操作符

// zip: 将两个流的第 N 个事件配对
// 适合：两个流严格配对（如请求和响应）
Rx.zip2(streamA, streamB, (a, b) => Pair(a, b));

// merge: 合并多个流为一个流，保持事件顺序
// 适合：将多个互斥的事件源合并（如多个按钮的点击事件）
Rx.merge([streamA, streamB, streamC]);

// switchMap: 每个事件都映射为一个新流，但只关心最新的流
// 适合：搜索——新搜索开始时，取消旧的搜索请求
searchQueryStream.switchMap((query) => searchAPI(query));
// ↑ 如果用户输入"A"→"AB"→"ABC"，前两次的 searchAPI 调用被自动取消

// concatMap: 类似于 switchMap，但不取消旧的流，而是排队
// 适合：必须按顺序处理的异步操作
uploadQueue.concatMap((item) => uploadItem(item));

// startWith: 在流的开头插入一个初始值
stream.startWith('initial value');

// sampleTime: 周期性采样
highFrequencyStream.sampleTime(const Duration(seconds: 1));

八、RxDart 增强：BehaviorSubject、ReplaySubject 与更多操作符

8.1 Subject 三种变体

RxDart 提供了三种 Subject，各有不同的缓存行为：

// 1. PublishSubject（默认 Subject）
// 不缓存，新订阅者只收到订阅之后的事件
final publishSub = PublishSubject<int>();
publishSub.add(1);
publishSub.add(2);
publishSub.listen(print);   // 不会收到 1 和 2
publishSub.add(3);          // 输出: 3

// 2. BehaviorSubject（带初始值/当前值的 Subject）
// 缓存最新值，新订阅者立即收到最新值
final behaviorSub = BehaviorSubject<int>.seeded(0);  // 初始值 0
behaviorSub.add(1);
behaviorSub.add(2);
behaviorSub.listen(print);  // 立即输出: 2（最新值）
behaviorSub.add(3);          // 输出: 3

// 获取当前值而不订阅
print(behaviorSub.value);    // 3

// 3. ReplaySubject（带历史缓存的 Subject）
// 缓存最近的 N 个事件，新订阅者回放这些事件
final replaySub = ReplaySubject<int>(maxSize: 3);
replaySub.add(1);
replaySub.add(2);
replaySub.add(3);
replaySub.add(4);
replaySub.listen(print);  // 立即输出: 2, 3, 4（最近 3 个）
replaySub.add(5);          // 输出: 5

8.2 在 E-Brufen 中的应用场景

Subject 类型E-Brufen 使用场景原因
PublishSubject用户操作事件（点击、滑动）新订阅者不需要历史事件
BehaviorSubject当前播放状态（正在播放/暂停）新页面需要知道当前状态
BehaviorSubject当前选中的白噪音场景定时器页面需要知道选择了哪个场景
ReplaySubject最近的错误日志（maxSize: 20）调试页面需要看到最近的错误
ReplaySubject最近的情绪记录变更（maxSize: 50）撤销功能需要历史

8.3 BehaviorSubject 实战：播放状态管理

在 E-Brufen 中，如果我们用 BehaviorSubject 替代当前的 StreamController：

class OhosAudioPlayerV2 {
  static final _playbackSubject = BehaviorSubject<PlaybackEvent>.seeded(
    const PlaybackEvent(PlaybackState.stopped),  // 初始状态：已停止
  );

  /// 播放状态变化流
  static Stream<PlaybackEvent> get onPlaybackStateChanged =>
      _playbackSubject.stream;

  /// 获取当前播放状态——无需等待下一个事件！
  static PlaybackState get currentState => _playbackSubject.value.state;

  static void _handleStateChange(PlaybackState newState) {
    _playbackSubject.add(PlaybackEvent(newState));
  }
}

// UI 层——新页面打开时立即知道当前状态
class NowPlayingBar extends StatefulWidget {
  @override
  State<NowPlayingBar> createState() => _NowPlayingBarState();
}

class _NowPlayingBarState extends State<NowPlayingBar> {
  @override
  void initState() {
    super.initState();
    // 新订阅者立即收到最新播放状态
    // 不需要额外的 "获取当前状态" 的 MethodChannel 调用
  }

  @override
  Widget build(BuildContext context) {
    return StreamBuilder<PlaybackEvent>(
      stream: OhosAudioPlayerV2.onPlaybackStateChanged,
      // initialData 可以省略——BehaviorSubject 自带初始值
      builder: (context, snapshot) {
        final isPlaying = snapshot.data?.state == PlaybackState.playing;
        return Icon(isPlaying ? Icons.pause : Icons.play_arrow);
      },
    );
  }
}

8.4 Dart 原生 Stream 与 RxDart Subject 对照

Dart 原生RxDart 等价新增能力
StreamController<T>()PublishSubject<T>()—
StreamController<T>.broadcast()PublishSubject<T>()—
—BehaviorSubject<T>.seeded(val)缓存最新值、.value 同步获取
—ReplaySubject<T>(maxSize: n)缓存历史 N 个值、回放给新订阅者
Stream.fromFuture(f)Stream.fromFuture(f)通用构造方式，无变化
Stream.periodic(d, cb)Stream.periodic(d, cb)通用构造方式，无变化

引入 RxDart 的核心收获不是"可以写出原本写不了的代码"，而是"用标准 API 替代手工实现的复杂模式"。

九、Stream 常见错误与避坑指南

9.1 错误一：单订阅流被多次监听

这是 Dart Stream 初学者最常遇到的运行时错误：

final controller = StreamController<int>();    // 默认：单订阅流
final stream = controller.stream;

stream.listen(print);   // ✅ 第一个监听者
stream.listen(print);   // ❌ StateError: Stream has already been listened to.

场景还原：在 build 方法中使用 StreamBuilder，Widget 重建时创建了新的订阅，而旧的订阅未被取消。

修复方案：

// 方案 A：使用广播流（适合多订阅者场景）
final controller = StreamController<int>.broadcast();

// 方案 B：使用 asBroadcastStream()
final broadcastStream = singleSubStream.asBroadcastStream();

// 方案 C：保证只有一个订阅者（在 initState 中 subscribe，在 dispose 中 cancel）

9.2 错误二：忘记取消订阅（内存泄漏）

class LeakyWidget extends StatefulWidget {
  @override
  _LeakyWidgetState createState() => _LeakyWidgetState();
}

class _LeakyWidgetState extends State<LeakyWidget> {
  @override
  void initState() {
    super.initState();
    someStream.listen((data) {
      setState(() { /* ... */ });   // ← 这个闭包捕获了 this
    });
    // ❌ 没有保存 StreamSubscription 引用
    // ❌ dispose 中无法 cancel
  }

  @override
  void dispose() {
    // ❌ 无法取消订阅！
    super.dispose();
  }
}

后果链：

StreamController（通常为静态/全局）
  └─ Stream
      └─ StreamSubscription
          └─ onData 回调（闭包）
              └─ 捕获 this（_LeakyWidgetState）
                  └─ 持有 context, AnimationController, Timer...
                      └─ 整个 Widget 树泄漏

修复：

class FixedWidget extends StatefulWidget {
  @override
  _FixedWidgetState createState() => _FixedWidgetState();
}

class _FixedWidgetState extends State<FixedWidget> {
  StreamSubscription? _sub;     // ✅ 保存引用

  @override
  void initState() {
    super.initState();
    _sub = someStream.listen((data) {
      if (!mounted) return;     // ✅ 安全检查
      setState(() { /* ... */ });
    });
  }

  @override
  void dispose() {
    _sub?.cancel();             // ✅ 取消订阅
    super.dispose();
  }
}

9.3 错误三：流的异常传播

Stream 中的未捕获异常会导致流关闭，且不会传播到外部的 try-catch：

final controller = StreamController<int>();

controller.stream
    .map((n) {
      if (n == 42) throw Exception('不能是 42');
      return n * 2;
    })
    .listen(
      (data) => print(data),
      onError: (error) => print('捕获错误: $error'),  // ✅ 必须处理
      onDone: () => print('流已关闭（可能因为未处理的错误）'),
    );

controller.add(1);   // 输出:  2
controller.add(42);  // 输出:  捕获错误: Exception: 不能是 42
                     //       流已关闭（可能因为未处理的错误）
controller.add(3);   // ❌ 不会输出任何东西——流已经在错误后关闭了

修复：在流的变换中主动处理错误，避免流提前关闭：

controller.stream
    .map((n) {
      if (n == 42) throw Exception('不能是 42');
      return n * 2;
    })
    .handleError((error) => print('跳过错误数据'))
    .listen(print);

9.4 错误四：在 build 中创建 StreamController

@override
Widget build(BuildContext context) {
  final controller = StreamController<int>(); // ❌ 每次 build 都创建新的！
  return StreamBuilder<int>(
    stream: controller.stream,
    builder: (context, snapshot) => Text('${snapshot.data}'),
  );
}

每次 build 都创建一个新的 StreamController，旧的 Controller 没有被关闭——Controller 泄漏。同时 StreamBuilder 订阅了旧的流，但老的 Controller 已经无人引用了（只有 StreamBuilder 还在等待它的数据）。

修复：在 initState 中创建 Controller，在 dispose 中关闭。

9.5 Stream 错误速查表

错误症状编译错误？修复
单订阅流多次监听StateError: Stream already listened to否（运行时）使用 broadcast() 或 asBroadcastStream
未取消订阅内存持续增长，页面退出后 setState 报错否dispose 中 cancel
StreamController 未关闭内存泄漏否dispose 中 close
流中异常未处理流提前关闭否用 handleError 或 onError 回调
build 中创建 Controller多次创建，旧的泄漏否在 initState 创建
add 在 close 之后StateError: Cannot add event after closing否检查 isClosed 属性
pause 后事件堆积内存占用异常增大否不要长时间 pause 高频流

十、Stream vs ChangeNotifier vs Future：三大响应式方案对比

10.1 语义对比

Future<Data>      —— "请在未来某个时刻给我一个 Data"
                    一次性、不可取消、成功或失败

ChangeNotifier    —— "我变了，你来查"
                    Pull 模式、无数据携带、简单直观

Stream<Event>     —— "我会持续发送 Event，直到我说结束"
                    Push 模式、有数据携带、可取消、可变换

10.2 核心技术维度对比

维度FutureChangeNotifierStream
值数量10（仅通知）0 到 N
数据传输有（携带结果/错误）无（仅信号）有（携带数据）
取消能力否解绑监听cancel()
变换操作.then().catchError()无.map().where().transform()
组合操作Future.wait()手动协调combineLatest, merge, zip
背压处理N/AN/Apause/resume
内置 WidgetFutureBuilderListenableBuilderStreamBuilder
外部依赖000（Dart 原生）
学习曲线低低中
适合场景HTTP请求、DB查询、单次操作UI 状态（简单的数据容器）事件流、实时数据、原生通信

10.3 场景决策矩阵

需求：数据会随时间多次变化吗？
  │
  ├── 否 ──► 数据是一次性的吗？
  │            │
  │            ├── 是 ──► Future
  │            │         例：HTTP 请求、文件读取、数据库查询
  │            │
  │            └── 否 ──► 同步代码即可
  │                      例：配置文件读取、常量定义
  │
  └── 是 ──► 变化是"事件"还是"状态"？
               │
               ├── 事件（离散、不可重复）──► Stream
               │   例：用户点击、传感器数据、WebSocket 消息
               │
               └── 状态（持续、可查询）──► 是否需要传递变更详情？
                                            │
                                            ├── 不需要 ──► ChangeNotifier
                                            │   例：计数器、开关状态、主题切换
                                            │
                                            └── 需要 ──► Stream<Event>
                                                例：数据库CRUD、文件系统变更

10.4 E-Brufen 中的实际选型

E-Brufen 场景当前方案推荐方案理由
情绪数据 CRUD 变更通知ChangeNotifierStream需要区分操作类型，支持增量更新
播放状态同步（原生→Dart）StreamStream已经是 Stream，无需改变
音频播放（Dart→原生）Future (MethodChannel)Future一次性操作，返回播放结果
设置读取同步 getterFuture（如果需要异步）当前同步读取 Hive 足够，无需改变
搜索输入尚未实现Stream（防抖）需要防抖，Stream 是最佳方案
呼吸动画计时Timer（在 BreathingCircle 内部）Timer 足够本地计时，不需要跨组件共享
统计数据计算每次调用重新计算combineLatest（两个数据源）需要响应情绪数据和日期范围的变化

10.5 渐进式引入 Stream 的路线图

对于类似 E-Brufen 的中小规模 Flutter 应用，建议采用渐进式引入策略：

第 1 步：在需要"事件传递"的新功能中使用 Stream
         （如：原生通信、WebSocket、传感器数据）
         │
第 2 步：将已有的"通知型"ChangeNotifier 添加 Stream 出口
         （双轨运行——同时保留 ChangeNotifier 和 Stream）
         │
第 3 步：将高频更新的 UI 组件从 ChangeNotifier 迁移到 Stream
         （增量更新取代全量刷新）
         │
第 4 步：当 Stream 操作足够复杂时，引入 RxDart
         （combineLatest、switchMap、debounceTime 等）

关键原则：不要为了"使用 Stream"而使用 Stream。ChangeNotifier 在简单的 UI 状态管理场景下已经足够好。Stream 的价值体现在：多消费者、事件携带数据、需要变换操作、需要组合多个数据源时。

十一、总结

核心观点

  1. Stream 是 Dart 原生的响应式通道——它不只是一个语言特性，更是连接"数据生产者"和"数据消费者"的标准管道。在 E-Brufen 中，Stream 真实地连接了鸿蒙原生端和 Flutter UI 端。
  2. 单订阅流和广播流的选择影响全局架构——单订阅流适合一次性的数据传递，广播流适合事件总线。E-Brufen 的 onPlaybackStateChanged 选择广播流，因为它天然会有多个订阅者。
  3. Stream 的变换操作（map、where、transform）构建了数据处理管道——将复杂的数据处理逻辑分解为多个独立的变换步骤，每一步都可测试、可替换。
  4. ChangeNotifier + Stream 双轨运行是渐进式迁移的最佳策略——不需要一次性推翻现有代码。先在 ChangeNotifier 旁边加上 Stream 出口，让新旧 UI 各自选择消费方式。
  5. RxDart 的价值在于"用标准 API 替代手工模式"——debounceTime 替代手写 Timer、combineLatest 替代手动协调、switchMap 替代手动取消旧请求。它不是必需品，但能大幅减少响应式编程的样板代码。
  6. Stream 的内存安全依赖手动管理——cancel() 和 close() 是你的责任。忘记它们，StreamController 和 StreamSubscription 会成为难以排查的内存泄漏源。

关键文件索引

文件Stream 相关代码
lib/data/ohos_audio.dartStreamController<PlaybackEvent>.broadcast() — 原生→Dart 事件流
lib/pages/soundscape/soundscape_page.dart_playbackSub.listen() — UI 订阅播放状态
lib/pages/breathe/breathe_page.dartTimer.periodic — 虽然不是 Stream 但思想相近
lib/data/mood_storage.dart可以改造为 Stream<MoodEvent> 出口

一句话总结

ChangeNotifier 告诉你"数据变了"，Stream 告诉你"数据变成了什么"。对于简单的 UI 刷新，前者足够；对于需要追踪变更细节、组合多个数据源、构建处理管道的场景，Stream 是 Dart 给出的原声答案——不依赖第三方库，鸿蒙平台完全支持，且与 Flutter 的 StreamBuilder 无缝集成。

作者简介：E-Brufen Dev，Flutter & 鸿蒙开发者，专注于跨平台移动应用开发与心理健康数字化。E-Brufen 是 AtomGit 上的开源 Flutter 鸿蒙应用，使用纯 Hive + ChangeNotifier 架构，零原生依赖，E-Brufen 项目。

# flutter # harmonyos # 华为 # android # 观察者模式 # 音视频
Logo
人工智能6S服务平台

作为“人工智能6S店”的官方数字引擎，为AI开发者与企业提供一个覆盖软硬件全栈、一站式门户。

更多推荐

cover

给鸿蒙 App 增加 JSON 驱动动态组件渲染能力，用 JSON 描述界面、状态驱动实时更新、模板变量自动绑定，纯 Dart 零权限六端通用 —— json_component 鸿蒙使用指南

avatar 人工智能6S服务平台
cover

HarmonyOS鸿蒙App 增加可折叠 JSON 树形浏览器，支持任意嵌套结构展开收起、自动展开层级控制、自定义节点渲染回调，纯 Dart 零权限即插即用 —— json_viewer 鸿蒙使用指南

avatar 人工智能6S服务平台
cover

Flutter动态JSON组件鸿蒙适配实战：json_component纯Dart能力兼容改造与全场景功能落地

avatar 人工智能6S服务平台