如何解决Dart流订阅中listen()调用频率的抖动问题？

问题描述 #

我在Dart/Flutter项目中使用ZMQ库搭建PUB/SUB TCP套接字通信，发布端每100ms推送一条消息。接收端采用以下代码监听消息：

m_frameDataSubscripiton = m_subscribeSocket.messages.listen((msg) async {
    Iterator<ZFrame> it = msg.iterator;
    if (it.moveNext()) {
      while (it.current.hasMore) {
        it.moveNext();
      }
      if (it.current.payload.isNotEmpty) {
        print("Got the frame packet at " + DateTime.now().millisecondsSinceEpoch.toString() + " size " + it.current.payload.length.toString());
      }
    }
  },
  onError: (Object error) {
    m_status = "subscription error";
    notifyListeners();
  },
  onDone: () {
    m_status = "done";
    notifyListeners();
  });

消息能正常接收，但listen回调触发呈突发状（间隔1-2ms连续处理多条），随后停顿约1000ms。尝试过降低接收套接字高水位（低于默认值1000会报错）、使用RxDart定时缓冲区（仍每1000ms轮询且存在消息丢失），均未解决问题。希望实现订阅端收到单条消息就立即处理，消除这种突发停顿现象。

解决方案 #

• 调整ZMQ套接字接收超时参数
  ZMQ的SUB套接字默认可能采用了批量缓冲策略，可通过设置ZMQ_RCVTIMEO选项强制套接字尽快返回可用消息，避免攒批等待。在Dart ZMQ库中通过setOption方法配置：

  m_subscribeSocket.setOption(ZMQ_RCVTIMEO, 1);
  

  将超时设为1ms，套接字会在无消息时立即返回，确保新消息到达时能触发回调及时处理。

• 简化回调内的异步逻辑
  当前回调标记为async但未使用await，会引入不必要的事件循环调度延迟，建议移除async：

  m_frameDataSubscripiton = m_subscribeSocket.messages.listen((msg) {
    Iterator<ZFrame> it = msg.iterator;
    if (it.moveNext()) {
      while (it.current.hasMore) {
        it.moveNext();
      }
      if (it.current.payload.isNotEmpty) {
        print("Got the frame packet at ${DateTime.now().millisecondsSinceEpoch} size ${it.current.payload.length}");
      }
    }
  },
  onError: (Object error) {
    m_status = "subscription error";
    notifyListeners();
  },
  onDone: () {
    m_status = "done";
    notifyListeners();
  });
  

• 排查发布端推送逻辑
  确认发布端是严格每100ms单条推送，无批量发送或事件循环阻塞导致的消息堆积情况。比如检查发布端定时器是否正常工作，是否存在耗时操作阻塞推送流程。

• 避免接收端事件循环阻塞
  若消息处理逻辑（除print外）存在耗时操作，需将其移至单独Isolate执行，防止阻塞Dart事件循环导致消息处理延迟，表现为突发停顿。

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