1
import 'dart:async';
2

3
void main() async {
4
  final controller = StreamController<int>();
5

6
  Stream<int> stream = controller.stream;
7
  StreamSink<int> sink = controller.sink;
8

9
  StreamSubscription<int> subscription = stream.listen(
10
    (data) => print('接收到数据: $data'),
11
    onError: (error) => print('错误: $error'),
12
    onDone: () => print('流已关闭'),
13
  );
14

15
  sink.add(1);
16
  sink.add(2);
17
  sink.add(3);
18

19
  await Future.delayed(Duration(seconds: 1));
20

21
  sink.addError('发生错误');
22

23
  await controller.close();
24
  await subscription.cancel();
25
}

1
import 'dart:async';
2

3
Stream<int> countStream(int to) async* {
4
  for (int i = 1; i <= to; i++) {
5
    yield i;
6
  }
7
}
8

9
void main() async {
10
  final controller = StreamController<int>.broadcast();
11

12
  Stream<int> broadcastStream = controller.stream;
13

14
  StreamSubscription<int> subscription1 = broadcastStream.listen(
15
    (data) => print('订阅1: $data'),
16
    onDone: () => print('订阅1完成'),
17
  );
18
  StreamSubscription<int> subscription2 = broadcastStream.listen(
19
    (data) => print('订阅2: $data'),
20
    onDone: () => print('订阅2完成'),
21
  );
22

23
  controller.add(1);
24
  controller.add(2);
25

26
  Stream<int> countToFive = countStream(5);
27
  await for (final count in countToFive) {
28
    controller.add(count);
29
  }
30

31
  subscription1.pause();
32

33
  controller.add(6);
34
  controller.add(7);
35

36
  subscription1.resume();
37

38
  await controller.close();
39

40
  await subscription1.cancel();
41
  await subscription2.cancel();
42
}