无论如何，是不是有一个新的流不会每次都检查测试方法？

Posted 2023-03-07

【中文标题】无论如何，是不是有一个新的流不会每次都检查测试方法？

【英文标题】：Is there anyway to derive a new stream that doesn't check the test method every time?无论如何，是否有一个新的流不会每次都检查测试方法？

【发布时间】：2014-10-30 18:57:57

【问题描述】：

我正在寻找一种基于测试方法从广播流中派生新流的方法，但这不会导致每个订阅都调用测试方法：

Stream where(Function bool test(T event))

从此流创建一个丢弃一些数据事件的新流。

新流发送与此流相同的错误和完成事件，但它只发送满足测试的数据事件。

如果此流是，则返回的流是广播流。如果多次收听广播流，则每个订阅将 单独执行测试。

是否有任何其他方法可以从源广播流派生子广播流，不会为每个订阅调用测试方法？

【参考方案1】：

Quiver 的StreamRouter 可能并不完全符合您的要求，但它只对每个事件执行一次测试，而不是对每个订阅者执行一次测试。它有点类似于您的Emitter，只是事件只能流入一个派生流。

var router = new StreamRouter(new Stream.fromIterable(range(0, 10)));
var odds = router.route((i) => i % 2 == 1);
odds.listen((i)  print(i); );
odds.listen((i)  print("$i is odd"); );

传递给route 的谓词每个事件只调用一次。

这是文档：

http://www.dartdocs.org/documentation/quiver/0.19.0-dev.3/index.html#quiver/quiver-async.StreamRouter

另外，并不是说您的 Emmiter 不需要缓存 Streams 来获得仅调用一次测试的行为，这来自手动创建控制器并调用 add()。所以你可以得到同样的东西：

/// Get the stream of [type].
Stream on(Stream source, Type type) 
  var controller = new StreamController();
  controller.addStream(source.where(_typeMatcher(type)));
  return controller.stream;

【讨论】：

缓存是必需的，就像您不存储新的控制器流一样，然后一遍又一遍地调用发射器.on(type) 将继续创建新的流并再次为每个订阅的事件强制执行谓词测试方法.谢谢你的颤抖，它当然值得考虑。

缓存可以防止对on(type)的多次调用导致多次测试调用，但是如果你调用on(source, type)一次，然后在结果上多次调用listen()，你不会得到缓存每个订户进行多次测试。所以这是多个listen() 调用或多个on(type) 调用的问题。关于缓存的一件事是您需要可以正确缓存的测试，闭包并非总是如此，尽管Type会如此。

【参考方案2】：

我在核心异步 API 中找不到简单的解决方案，但我发现如果您将流转发到每种类型的测试的另一个流，性能节省是巨大的，this 是我的班级使用这个流转发技术，它比下面的第二个原始版本执行得要好得多，它将强制对每个发射对象的每个订阅进行类型检查：

/// A mixin class to enable any class to act as a generic emitter of objects.
class Emitter 

  final StreamController _controller = new StreamController.broadcast();
  final Map<Type, Stream> _streamCache = new Map<Type, Stream>();

  /**
  * Emit an object.
  *
  *     emit(new Foo());
  *
  * Will send the new Foo object down the stream returned by `on(Foo)`.
  */
  void emit(obj) => _controller.add(obj);

  /// Get the stream of [type].
  Stream on(Type type)
    var stream = _streamCache[type];
    if(stream == null)
      StreamController controller = new StreamController.broadcast();
      _streamCache[type] = stream = type == All? _controller.stream: controller.stream;
      if(type != All)
        _controller.stream.where(_typeMatcher(type)).listen(controller.add, onError: controller.addError, onDone: controller.close);
      
    
    return stream;
  

这是 on 方法的旧的非常慢的实现：

  /// Get the stream of [type].
  Stream on(Type type) => type == All? _controller.stream:  _controller.stream.where(_typeMatcher(type));

对于某些性能数据，如果您向 200 个侦听器发出 100000 个对象，则高性能运行时间为 6.6 秒，慢速版本运行时间为 20.2 秒。