Dart Stream

Stream ist der Mechanismus in Dart für den Umgang mit Sequenzen kontinuierlicher asynchroner Ereignisse.

Wenn Future ein einmaliges asynchrones Ergebnis ist, dann ist Stream ein wiederholter, kontinuierlicher asynchroner Datenfluss.

Dieses Kapitel stellt das Konzept von Stream, await for zum Abhören, die Erstellung von StreamController und die Unterschiede zwischen Single-Subscription- und Broadcast-Streams vor.

Streams und Ereignissequenzen

Ein Stream ist wie ein Förderband; die Daten kommen im Laufe der Zeit eins nach dem anderen an.

Du musst nicht auf alle Daten auf einmal warten; stattdessen verarbeitest du jedes Stück, sobald es eintrifft.

Typische Anwendungsfälle für Streams:

Szenarien	Beispiel
Benutzereingabe-Ereignisse	Button-Klicks, Mausbewegung, Tastatureingabe
Dateilesen	Große Dateien Zeile für Zeile lesen
Netzwerkdaten	WebSocket-Nachrichten, Echtzeit-APIs
Timer	Timer-Ereignisse, die einmal pro Sekunde ausgelöst werden
Zustandsänderungen	Zustandsverwaltungs-Streams in Flutter.

Beispiel

Grundlegende Verwendung von Stream — Zuhören mit listen:

void main() {
// Erstelle einen Stream: Gib jede 1 Sekunde eine Zahl aus
var stream = Stream<int>.periodic(
Duration(seconds: 1),
(count) => count + 1, // count beginnt bei 0
);

print('Beginne, auf den Stream zu hören...');

// listen abonniert den Stream
var subscription = stream.listen(
(data) {
// Wird aufgerufen, sobald neue Daten eintreffen
print('BEISPIEL empfangene Daten: $data');
},
onError: (error) {
// Wird aufgerufen, wenn Stream auf einen Fehler stößt
print('Fehler: $error');
},
onDone: () {
// Wird aufgerufen, wenn Stream geschlossen wird
print('Stream geschlossen');
},
);

// Abonnement nach 5 Sekunden abbrechen
Future.delayed(Duration(seconds: 5), () {
subscription.cancel();
print('Abonnement abgebrochen');
});
}

开始监听 Stream...
EXAMPLE 收到数据: 1
EXAMPLE 收到数据: 2
EXAMPLE 收到数据: 3
EXAMPLE 收到数据: 4
EXAMPLE 收到数据: 5
已取消订阅

listen() gibt ein StreamSubscription-Objekt zurück, mit dem du das Abonnement steuern kannst:

Methoden	Funktion
subscription.pause()	Datenempfang pausieren
subscription.resume()	Datenempfang fortsetzen
subscription.cancel()	Abonnement abbrechen
subscription.isPaused	Ob es sich in einem pausierten Zustand befindet

await for zum Abhören

Neben dem Callback-Ansatz mit listen() kannst du auch eine await for-Schleife verwenden, um einen Stream zu konsumieren.

await for macht die Logik der Verarbeitung eines Streams so klar wie eine gewöhnliche for-Schleife.

Beispiel

// Erzeuge einen endlichen Datenstrom
Stream<int> countStream(int max) async* {
for (int i = 1; i <= max; i++) {
await Future.delayed(Duration(milliseconds: 300));
yield i; // yield gibt Daten an den Stream aus
}
}

Future<void> main() async {
print('Beginne mit dem Konsumieren des Streams mit await for...');

// await for: auf jedes eintreffende Datenelement warten und sie nacheinander verarbeiten
await for (var value in countStream(5)) {
print('BEISPIEL Anzahl: $value');
}

print('Konsumieren des Streams abgeschlossen');
}

开始用 await for 消费 Stream...
EXAMPLE 计数: 1
EXAMPLE 计数: 2
EXAMPLE 计数: 3
EXAMPLE 计数: 4
EXAMPLE 计数: 5
Stream 消费完毕

Eigenschaften von await for:

Die Syntax ähnelt einer normalen for-Schleife, aber jede Iteration wartet darauf, dass das nächste Datenelement eintrifft.
Wenn der Stream geschlossen wird, endet die Schleife automatisch.
Es kann nur in asynchronen Funktionen verwendet werden.
Wenn du vorzeitig aussteigen musst, verwende break (genau wie bei einer normalen for-Schleife).

await for eignet sich für Szenarien, in denen du alle Daten sequenziell verarbeiten musst, z. B. beim Lesen aller Zeilen aus einer Datei. listen() eignet sich für Szenarien, in denen du kontinuierlich auf Ereignisse reagieren musst, z. B. bei Button-Klicks. Die beiden können einander ersetzen, haben aber jeweils ihre eigenen Stärken.

StreamController erstellt einen Stream

StreamController ist ein Werkzeug zum manuellen Erstellen und Steuern von Streams.

Du kannst jederzeit Daten oder Fehler hinzufügen oder ihn schließen.

Beispiel

import 'dart:async';

// Verwende StreamController, um einen einfachen Countdown-Timer zu implementieren
class CountdownTimer {
final StreamController<int> _controller = StreamController<int>();
Timer? _timer;
int _remaining = 0;

// Stream für externes Abonnement bereitstellen
Stream<int> get tickStream => _controller.stream;

// Countdown starten
void start(int seconds) {
_remaining = seconds;

// Anfangswert sofort senden
_controller.add(_remaining);

_timer = Timer.periodic(Duration(seconds: 1), (timer) {
_remaining--;

if (_remaining > 0) {
_controller.add(_remaining); // Daten senden
} else {
_controller.add(0); // Die abschließende 0 senden
_controller.close(); // Stream schließen
timer.cancel();
}
});
}

// Countdown abbrechen
void cancel() {
_timer?.cancel();
_controller.addError('Countdown abgebrochen'); // Fehler senden
_controller.close();
}

// Ressourcen freigeben
void dispose() {
_controller.close();
}
}

Future<void> main() async {
var timer = CountdownTimer();

// Countdown-Event abonnieren
timer.tickStream.listen(
(remaining) {
print('BEISPIEL Countdown: $remaining Sekunden');
},
onError: (error) {
print('Fehler: $error');
},
onDone: () {
print('Countdown beendet!');
},
);

timer.start(5);

// Warten, bis der Countdown abgeschlossen ist
await Future.delayed(Duration(seconds: 6));
timer.dispose();
}

EXAMPLE 倒计时: 5 秒
EXAMPLE 倒计时: 4 秒
EXAMPLE 倒计时: 3 秒
EXAMPLE 倒计时: 2 秒
EXAMPLE 倒计时: 1 秒
EXAMPLE 倒计时: 0 秒
倒计时结束！

async* Generatorfunktionen

Wenn du nur einfach eine Sequenz von Daten erzeugen musst, ist async* praktischer als StreamController.

Beispiel

// async* kennzeichnet dies als asynchrone Generatorfunktion
// Der Rückgabetyp muss Stream sein
Stream<String> readLinesAsync() async* {
var lines = ['Erste Zeile', 'Zweite Zeile', 'Dritte Zeile', 'BEISPIEL'];

for (var line in lines) {
await Future.delayed(Duration(milliseconds: 500));
yield line; // yield gibt Daten an den Stream aus
}
// Der Stream wird automatisch geschlossen, wenn die Funktion endet
}

// Asynchroner Generator mit Fehlerbehandlung
Stream<int> generateNumbersWithError() async* {
for (int i = 1; i <= 5; i++) {
await Future.delayed(Duration(milliseconds: 300));

if (i == 3) {
throw Exception('Nummer 3 hatte einen Fehler!');
}
yield i;
}
}

Future<void> main() async {
print('Zeile für Zeile lesen:');
await for (var line in readLinesAsync()) {
print(' $line');
}

print('\n'Generator mit Fehlern:');
try {
await for (var num in generateNumbersWithError()) {
print(' Number: $num');
}
} catch (e) {
print(' Caught error: $e');
}
}

逐行读取：
  第一行
  第二行
  第三行
  EXAMPLE

带错误的生成器：
  数字: 1
  数字: 2
  捕获错误: Exception: 数字 3 出错了！

Single-Subscription-Streams vs. Broadcast-Streams

Dart-Streams sind in zwei Typen unterteilt: Single-Subscription-Streams und Broadcast-Streams.

Single-Subscription-Stream

Dies ist der Standard-Stream-Typ.

Er kann nur von einem Listener abonniert werden, geeignet für Szenarien, in denen Sie ihn einmal von Anfang bis Ende konsumieren.

Beispiel

void main() {
// Single-Subscription-Stream (Standard)
var stream = Stream.fromIterable([1, 2, 3]);

// Erstes Abonnement: OK
stream.listen((data) => print('Subscriber 1: $data'));

// Zweites Abonnement: Fehler! Ein Single-Subscription-Stream kann nicht mehrfach abonniert werden
// stream.listen((data) => print('Abonnent 2: $data')); // Laufzeitfehler
}

Broadcast-Stream

Broadcast-Streams ermöglichen es mehreren Listenern, gleichzeitig zu abonnieren, und eignen sich für Event-Broadcast-Szenarien.

Beispiel

import 'dart:async';

void main() {
// Einen Broadcast-Stream erstellen
var controller = StreamController<int>.broadcast();

// Mehrere Abonnenten können gleichzeitig lauschen
controller.stream.listen(
(data) => print('BEISPIEL Abonnent A: $data empfangen'),
);

controller.stream.listen(
(data) => print('BEISPIEL Abonnent B: $data empfangen'),
);

// Daten senden, beide Abonnenten werden sie empfangen
controller.add(1);
controller.add(2);
controller.add(3);

// Ein spätes Abonnement kann weiterhin nachfolgende Daten empfangen (aber keine vorherigen Daten)
Future.delayed(Duration(seconds: 1), () {
controller.stream.listen(
(data) => print('BEISPIEL Später Abonnent C: $data empfangen'),
);
controller.add(4);
controller.close();
});
}

EXAMPLE 订阅者A: 收到 1
EXAMPLE 订阅者B: 收到 1
EXAMPLE 订阅者A: 收到 2
EXAMPLE 订阅者B: 收到 2
EXAMPLE 订阅者A: 收到 3
EXAMPLE 订阅者B: 收到 3
EXAMPLE 订阅者A: 收到 4
EXAMPLE 订阅者B: 收到 4
EXAMPLE 迟到的订阅者C: 收到 4

Vergleich der beiden Stream-Typen:

Funktionen	Single-Subscription-Stream	Broadcast-Stream
Anzahl der Abonnenten	Nur einer	Multiple
Daten-Replay	Von Anfang an konsumieren	Erhält Daten erst nach dem Abonnieren
Typische Szenarien	Dateilesen, HTTP-Antworten	Button-Klicks, Statusbenachrichtigungen
Erstellungsmethode	StreamController()	StreamController.broadcast()

Ein "später Abonnent" eines Broadcast-Streams kann Ereignisse erst nach dem Abonnieren empfangen; frühere Ereignisse wurden bereits verpasst. Dies ist der wichtigste Verhaltensunterschied zwischen Broadcast-Streams und Single-Subscription-Streams.

Gängige Stream-Transformationsmethoden

Stream bietet eine Reihe von Methoden zum Transformieren und Verarbeiten von Daten, ähnlich den funktionalen Methoden von List.

Beispiel

void main() async {
var numbers = Stream.fromIterable([1, 2, 3, 4, 5, 6]);

// map: jedes Datenelement transformieren
var doubled = numbers.map((n) => n * 2);
print('Doubled:');
await for (var n in doubled) {
print(' $n');
}

// where: Daten filtern
var source = Stream.fromIterable([10, 15, 20, 25, 30]);
var even = source.where((n) => n % 2 == 0);
print('Gerade:');
await for (var n in even) {
print(' $n');
}

// take: nur die ersten N Elemente nehmen
var infinite = Stream.periodic(
Duration(milliseconds: 100),
(i) => i + 1,
);
print('Nur die ersten 3 nehmen:');
await for (var n in infinite.take(3)) {
print(' BEISPIEL: $n');
}

// skip: die ersten N Elemente überspringen
var data = Stream.fromIterable([1, 2, 3, 4, 5]);
print('Die ersten 2 überspringen:');
await for (var n in data.skip(2)) {
print(' $n');
}

// distinct: Duplikate entfernen
var duplicates = Stream.fromIterable([1, 2, 2, 3, 3, 3]);
print('Distinct:');
await for (var n in duplicates.distinct()) {
print(' $n');
}
}

翻倍:
  2
  4
  6
  8
  10
  12
偶数:
  10
  20
  30
只取前 3 个:
  EXAMPLE: 1
  EXAMPLE: 2
  EXAMPLE: 3
跳过前 2 个:
  3
  4
  5
去重:
  1
  2
  3

Weitere Erweiterungen