Dart Stream 如何在下游消费过慢时丢弃事件（MJPEG解码场景）

问题根源不在标记的判断逻辑上，是被asyncMap的内置机制直接挡住了：

• Dart核心库的asyncMap是严格串行工作的：上游每发来一个事件，它会先把事件存进内部队列，等上一个通过map函数生成的Future完全执行完成、结果发送到下游之后，才会从队列里取出下一个事件执行map回调。
• 这意味着所有新到的JPEG帧，根本还没走到if (decoding)的判断代码行，就已经被asyncMap的内部队列存起来排队了。不管标记逻辑怎么写，队列里的帧只会越攒越多，最终导致延迟持续升高。

从根上绕开asyncMap的缓存机制，自己控制事件的分发节奏：上游事件进来第一时间就判断当前是否在处理上一个事件，如果是直接丢弃，既不触发处理逻辑，也不做任何队列缓存。

手写无缓存的自定义StreamTransformer #

这是最贴合MJPEG场景的方案，不需要依赖三方组件，核心就是把忙判断的逻辑放到事件进入管线的第一位置，不给排队留空间：

// 定义转换器：正在处理上一帧时，直接丢弃所有新传入的帧
StreamTransformer<List<int>, ui.Image> get dropFrameWhileBusy {
  // 解码状态存在闭包内部，避免多次调用listen时出现状态串扰
  bool isDecoding = false;
  return StreamTransformer.fromHandlers(
    handleData: (jpegFrame, sink) async {
      // 事件进来第一时间同步判断，不会等队列排队执行
      if (isDecoding) return;
      isDecoding = true;
      try {
        final decodedImage = await _decodeFrame(jpegFrame);
        // 只有解码完成且流未关闭时，才把结果发送给下游
        if (!sink.isClosed) sink.add(decodedImage);
      } catch (err) {
        if (!sink.isClosed) sink.addError(err);
      } finally {
        isDecoding = false;
      }
    },
  );
}

对应的业务代码只要把原来用asyncMap、做null过滤、类型转换的部分全部替换成这个转换器就行：

// 在transform(_mjpegSplitter)之后的逻辑替换为
.transform(dropFrameWhileBusy)
.listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError);

这个方案能生效的核心原因：

• StreamTransformer.fromHandlers的handleData回调是上游每发一个事件就立刻同步触发的，没有内置的事件缓存队列。
• 忙状态判断是在事件进入处理流程的第一行，只要上一帧还没解码完，新帧直接被丢弃，根本不会进入解码流程，也不会在内存里积压。

可替换方案：用“最新值优先”的流逻辑 #

如果不想手写转换器，也可以利用流的“取消旧订阅、只处理最新事件”的逻辑实现跳帧：
核心逻辑是上游每来一个新帧，就立刻取消上一个还在等待解码的流程，直接基于新帧发起解码。效果和上面的丢帧逻辑一致，永远只处理最新到达的帧，不会因为队列积压产生延迟。

注意实现细节：使用这个方案时要确保_decodeFrame支持取消操作——也就是运行解码逻辑的Isolate可以在收到取消信号时被销毁，避免出现主流程已经丢弃了旧帧、但后台Isolate还在空跑占用资源的问题。

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