时间戳

首页 > 解决方案 > RxDart shareReplay 不“热”？

dart - RxDart shareReplay 不“热”？

问题描述

我在我的应用程序中成功使用shareReplayfrom RxDart(0.24.1) 时遇到问题。我有一个Stream并且我想“缓存”最新发出的值，以便任何“迟到”的侦听器立即获取它（流可以空闲更长的时间）。我没有做到这一点，所以我开始在我的应用程序之外进行实验，并且必须承认我并不完全理解正在发生的事情。

这是一些代码：

import 'package:rxdart/rxdart.dart';

void main() async {
  final s = Stream.fromIterable([1, 2, 3]).shareReplay(maxSize: 10);
  print('First:');
  await for (final i in s) {
    print(i);
  }
  print('Second:');
  await for (final i in s) {
    print(i);
  }
}

我希望代码能够打印

First
1
2
3
Second
1
2
3

但它永远不会到达第二个await for，它甚至不打印字符串Second:；但是，程序成功完成，所以第一个await for确实完成了（因为原始Stream程序是有限的）。到底是怎么回事？

实际上，Stream我所拥有的可能是无穷无尽的，因为事件之间的时间很长），而且我只对最后一个事件感兴趣，所以这里有一些模拟这个的代码：

import 'package:rxdart/rxdart.dart';

void main() async {
  final s = Stream.periodic(Duration(seconds: 1), (e) => e).shareReplay(maxSize: 1);
  // print(s.isBroadcast);
  await Future.delayed(Duration(seconds: 3));
  print('First');
  await for (final i in s.take(1)) {
    print(i);
  }
  print('Second');
  await for (final i in s.take(1)) {
    print(i);
  }
}

为了确保Stream罐头完成（也await for可以完成），我使用take(1). 我预计输出是：

First
2
Second
2

（也许两个是三个？）我期望发生的是：

  1. 周期性流每秒开始发射（它是广播/热的）。
  2. 由于Future.delayed.
  3. 第一个迟到的听众来了，得到了最新的值，2（或者可能是 3？）并且由于take(1).
  4. 第二个迟到的侦听器也来了并获得了最新的值，因为在此期间流没有发出任何其他东西（即使它发出了，该值也会大 1，这没关系）并且由于take(1).
  5. 该应用程序完成。

但是，输出是：

First
0
Second
Unhandled exception:
type '_TakeStream<int>' is not a subtype of type 'Future<bool>'
#0      _StreamIterator._onData (dart:async/stream_impl.dart:1067:19)
#1      _RootZone.runUnaryGuarded (dart:async/zone.dart:1384:10)
#2      _BufferingStreamSubscription._sendData (dart:async/stream_impl.dart:357:11)
#3      _BufferingStreamSubscription._add (dart:async/stream_impl.dart:285:7)
#4      _ForwardingStreamSubscription._add (dart:async/stream_pipe.dart:127:11)
#5      _TakeStream._handleData (dart:async/stream_pipe.dart:318:12)
#6      _ForwardingStreamSubscription._handleData (dart:async/stream_pipe.dart:157:13)
#7      _RootZone.runUnaryGuarded (dart:async/zone.dart:1384:10)
#8      _BufferingStreamSubscription._sendData (dart:async/stream_impl.dart:357:11)
#9      _BufferingStreamSubscription._add (dart:async/stream_impl.dart:285:7)
#10     _SyncBroadcastStreamController._sendData (dart:async/broadcast_stream_controller.dart:385:25)
#11     _BroadcastStreamController.add (dart:async/broadcast_stream_controller.dart:250:5)
#12     _StartWithStreamSink._safeAddFirstEvent (package:rxdart/src/transformers/start_with.dart:56:12)
#13     _StartWithStreamSink.onListen (package:rxdart/src/transformers/start_with.dart:37:11)
#14     forwardStream.<anonymous closure>.<anonymous closure> (package:rxdart/src/utils/forwarding_stream.dart:31:37)
#15     forwardStream.runCatching (package:rxdart/src/utils/forwarding_stream.dart:24:12)
#16     forwardStream.<anonymous closure> (package:rxdart/src/utils/forwarding_stream.dart:31:16)
#17     _runGuarded (dart:async/stream_controller.dart:847:24)
#18     _BroadcastStreamController._subscribe (dart:async/broadcast_stream_controller.dart:213:7)
#19     _ControllerStream._createSubscription (dart:async/stream_controller.dart:860:19)
#20     _StreamImpl.listen (dart:async/stream_impl.dart:493:9)
#21     DeferStream.listen (package:rxdart/src/streams/defer.dart:37:18)
#22     StreamView.listen (dart:async/stream.dart:1871:20)
#23     new _ForwardingStreamSubscription (dart:async/stream_pipe.dart:118:10)
#24     new _StateStreamSubscription (dart:async/stream_pipe.dart:341:9)
#25     _TakeStream._createSubscription (dart:async/stream_pipe.dart:310:16)
#26     _ForwardingStream.listen (dart:async/stream_pipe.dart:83:12)
#27     _StreamIterator._initializeOrDone (dart:async/stream_impl.dart:1041:30)
#28     _StreamIterator.moveNext (dart:async/stream_impl.dart:1028:12)
#29     main (package:shopping_list/main.dart)
<asynchronous suspension>
#30     _startIsolate.<anonymous closure> (dart:isolate-patch/isolate_patch.dart:301:19)
#31     _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:168:12)

第一个await for真正take(1)使它完成的工作正如我所期望的那样，除了它获取初始值，这意味着流在侦听器到来之前没有发出任何状态（不是“热”）。第二个失败但有异常。我错过了什么？

编辑：我的理解shareReplay可能一直是错误的，我虽然很热，因为https://pub.dev/documentation/rxdart/latest/rx/ReplaySubject-class.html说它很热。

我稍微更改了代码：

import 'package:rxdart/rxdart.dart';

void main() async {
  final s = BehaviorSubject();
  s.addStream(Stream.periodic(Duration(seconds: 1), (e) => e));
  // print(s.isBroadcast);
  await Future.delayed(Duration(seconds: 3));
  print('First');
  await for (final i in s.take(1)) {
    print(i);
  }
  print('Second');
  await for (final i in s.take(1)) {
    print(i);
  }
}

现在输出是：

First
2
Second
2

但是程序永远不会完成......

没有take(1)输出永远不会到达第二个监听器，所以它确实有一些效果。我不明白。

编辑2：我想我可能知道为什么第一个示例shareReplay不起作用，这是文档的摘录：

它会在第一次收听时自动开始发射项目，并在没有收听者时关闭。在我的情况下，第一个侦听器/await for订阅，流开始，当循环完成时，最后一个（也是唯一的）侦听器完成，所以从返回的流shareReplay关闭。不过，仍然没有解释为什么该行print('Second');不执行。

标签： dartrxdart

解决方案

您的最终程序未终止的原因是您的广播流仍在每秒发出事件。没有人在听，但它是一个广播流，所以它不在乎。它有一个每秒钟都在滴答作响的计时器，该计时器使程序保持活力，即使没有人会再听一次。您也可以通过这样做来保持程序的活力Timer.periodic(Duration(seconds: 1), (_) {});。只要计时器处于活动状态，就可能发生某些事情。

不过，这次崩溃很有趣。它表明StreamIterator该类的内部逻辑存在问题。我会调查的。

不能说任何保证有用的东西shareReplay，我不熟悉 RxDart。我的猜测是第一个程序遇到问题是因为replayShared流没有正确终止——也就是说，done当它完成时没有发送事件。这可以解释程序提前终止的原因。第一个await for是等待下一个事件，所以它不会结束并进入下一个循环，但是没有任何类型的实时计时器或预定事件，所以隔离只是关闭。快速阅读BeheaviorSubject表明它像流控制器一样工作，因此您可能需要在addStream完成后关闭它。也许：

s.addStream(...).whenComplete(s.close);

推荐阅读