流相当于 Observable.Throttle？答案

【问题标题】：Streams equivalent to Observable.Throttle?流相当于 Observable.Throttle？
【发布时间】：2013-02-25 16:52:24
【问题描述】：

是否有等效于 Observable.Throttle 的 Streams？如果没有——是否有任何相当优雅的方式来实现类似的效果？

【问题讨论】：

标签： dart dart-async

【解决方案1】：

目前在流上没有这样的方法。已提交增强请求，您可以加注星标issue 8492。

但是，您可以使用 where 方法来做到这一点。在以下示例中，我定义了一个 ThrottleFilter 类来忽略给定持续时间内的事件：

import 'dart:async';

class ThrottleFilter<T> {
  DateTime lastEventDateTime = null;
  final Duration duration;

  ThrottleFilter(this.duration);

  bool call(T e) {
    final now = new DateTime.now();
    if (lastEventDateTime == null ||
        now.difference(lastEventDateTime) > duration) {
      lastEventDateTime = now;
      return true;
    }
    return false;
  }
}

main() {
  final sc = new StreamController<int>();
  final stream = sc.stream;

  // filter stream with ThrottleFilter
  stream.where(new ThrottleFilter<int>(const Duration(seconds: 10)).call)
    .listen(print);

  // send ints to stream every second, but ThrottleFilter will give only one int
  // every 10 sec.
  int i = 0;
  new Timer.repeating(const Duration(seconds:1), (t) { sc.add(i++); });
}

【讨论】：

【解决方案2】：

rate_limit package 提供流的节流和去抖动功能。

【讨论】：

与 Dart 2 不兼容。

【解决方案3】：

以下版本更接近于 Observable.Throttle 所做的：

class Throttle extends StreamEventTransformer {
  final duration;
  Timer lastTimer;

  Throttle(millis) :
    duration = new Duration(milliseconds : millis);


  void handleData(event, EventSink<int> sink) {
    if(lastTimer != null){
      lastTimer.cancel();
    }
    lastTimer = new Timer(duration, () => sink.add(event));
  }
}

main(){
  //...
  stream.transform(new Throttle(500)).listen((_) => print(_));
  //..
}

【讨论】：

【解决方案4】：

@Victor Savkin 的回答很好，但我总是尽量避免重新发明轮子。所以除非你真的只需要那个油门，否则我建议使用RxDart 包。由于您正在处理 Streams 和其他反应性对象，RxDart 除了节流之外还有很多不错的功能。

我们可以通过几种方式实现 500 毫秒的节流：

throttleTime 来自 ThrottleExtensions<T> Stream<T> 扩展名：stream.throttleTime(Duration(milliseconds: 500)).listen(print);
结合ThrottleStreamTransformer 与TimerStream: stream.transform(ThrottleStreamTransformer((_) => TimerStream(true, const Duration(milliseconds: 500)))).listen(print);
使用Debounce Extensions/DebounceStreamTransformer:stream.debounceTime(Duration(milliseconds: 500)).listen(print);

在延迟方面存在一些细微的差异，但它们都会受到限制。关于throttleTime 与debounceTime 的示例，请参见What is the difference between throttleTime vs debounceTime in RxJS and when to choose which?

【讨论】：