Dart: Dart Stream

Last updated: 2026-08-26

Stream is a river of data — continuous, processed as it arrives, no need to wait for everything to arrive at once.

1. What You Will Learn

Stream concepts: Single-subscription vs Broadcast
Stream creation: fromIterable / periodic / event-transformed
Operators: map / where / expand / take / skip / distinct
StreamController and StreamSubscription
Bob's Scenario: DataPipeline real-time order stream processing

2. A Developer's Real Story

(1) Pain Point: Millions of Data Records Cannot Be Loaded into Memory All at Once

Bob's DataPipeline initially loaded all 1,200,000 orders into memory for processing when handling e-commerce orders, resulting in a memory peak of 4GB and causing the server to crash with an OOM error three times. Furthermore, as real-time orders continued to arrive, the batch processing model was unable to handle the newly incoming data.

(2) Stream Solution

Stream allows data to be processed piece by piece like a water flow: process one as it arrives, keeping memory usage stable at 50MB. Stream operators (where/map/reduce) enable declarative definition of the processing pipeline.

DART

orderStream
  .where((order) => order.status == 'completed')
  .map((order) => order.amount)
  .fold(0, (sum, amount) => sum + amount)
  .then((total) => print('Revenue: \$$total USD'));

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

(3) Benefits

Memory reduced from 4GB to 50MB, eliminating OOM errors
Real-time processing of newly arriving orders, latency reduced from "waiting for batch processing" to "seconds"
Stream operator chains make data processing logic declarative and composable

3. Stream Basics

(1) Single-subscription vs Broadcast

flowchart LR
  A[Data Source] --> B["StreamController"]
  B --> C["where() Filter"]
  C --> D["map() Transform"]
  D --> E["expand() Flatten"]
  E --> F["take() Limit"]
  F --> G["listen() Consume"]
  subgraph Stream Pipeline
    C
    D
    E
    F
  end

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

Type	Subscribers	Replay	Use Case
Single-subscription Stream	1	Not replayable	File reading, HTTP responses
Broadcast Stream	Multiple	Not replayable	UI events, real-time data

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：Stream Basics

DART

import 'dart:async';

void main() async {
  // Create stream from iterable
  final stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  // Listen to stream
  await for (final value in stream) {
    print('Received: $value');
  }

  // Alternative: listen with callback
  final stream2 = Stream.fromIterable(['ORD-001', 'ORD-002', 'ORD-003']);
  stream2.listen(
    (order) => print('Order: $order'),
    onDone: () => print('Stream completed'),
    onError: (e) => print('Error: $e'),
  );

  // Wait for stream to complete
  await stream2.drain();
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

4. Stream Creation

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：fromIterable Creation

DART

import 'dart:async';

void main() async {
  // From iterable - emits each element
  final orders = Stream.fromIterable([
    {'id': 'ORD-001', 'amount': 1500.0},
    {'id': 'ORD-002', 'amount': 3200.0},
    {'id': 'ORD-003', 'amount': 890.0},
  ]);

  await for (final order in orders) {
    print('Order: ${order['id']} - \$${order['amount']} USD');
  }
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：periodic Creation

DART

import 'dart:async';

void main() async {
  // Periodic stream - emits values at regular intervals
  final ticker = Stream.periodic(
    const Duration(seconds: 1),
    (count) => 'Tick ${count + 1}',
  );

  // Take only first 5 ticks
  await for (final tick in ticker.take(5)) {
    print(tick);
  }
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：StreamController Creation

DART

import 'dart:async';

void main() async {
  final controller = StreamController``<String>``();

  // Add data to the stream
  controller.add('ORD-001');
  controller.add('ORD-002');
  controller.add('ORD-003');

  // Close the stream when done
  controller.close();

  // Consume the stream
  await for (final order in controller.stream) {
    print('Processing: $order');
  }
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

Creation Method	Syntax	Data Source	Use Case
fromIterable	`Stream.fromIterable(list)`	Existing collection	Testing, small datasets
periodic	`Stream.periodic(duration, fn)`	Timed generation	Timers, polling
StreamController	StreamController``<T>``()	Manually added	Event sources, real-time data
empty	`Stream.empty()`	None	Empty stream
value	`Stream.value(v)`	Single value	Wrapping a single value

5. Stream Operators

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：where — Filtering

DART

import 'dart:async';

void main() async {
  final amounts = Stream.fromIterable([1500.0, 50.0, 3200.0, 890.0, -100.0]);

  // Filter only positive amounts >= 100
  final highValue = amounts.where((a) => a >= 100);

  await for (final amount in highValue) {
    print('\$${amount.toStringAsFixed(2)} USD');
  }
  // Output: $1500.00 USD, $3200.00 USD, $890.00 USD
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：map — Transformation

DART

import 'dart:async';

void main() async {
  final amounts = Stream.fromIterable([1500.0, 3200.0, 890.0]);

  // Transform amounts to formatted strings
  final formatted = amounts.map((a) => '\$${a.toStringAsFixed(2)} USD');

  await for (final text in formatted) {
    print(text);
  }
  // Output: $1500.00 USD, $3200.00 USD, $890.00 USD
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：expand — Flattening

DART

import 'dart:async';

void main() async {
  final orders = Stream.fromIterable([
    {'id': 'ORD-001', 'items': ['Laptop', 'Mouse']},
    {'id': 'ORD-002', 'items': ['Keyboard']},
  ]);

  // Expand: each order → multiple items
  final items = orders.expand((order) =>
      (order['items'] as List).cast``<String>``());

  await for (final item in items) {
    print('Item: $item');
  }
  // Output: Item: Laptop, Item: Mouse, Item: Keyboard
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：take / skip / distinct

DART

import 'dart:async';

void main() async {
  final data = Stream.fromIterable([1, 1, 2, 2, 3, 3, 4, 5]);

  // Take first 3 elements
  print('take(3):');
  await data.take(3).forEach(print);  // 1, 1, 2

  // Skip first 2
  print('skip(2):');
  await Stream.fromIterable([1, 1, 2, 2, 3]).skip(2).forEach(print);  // 2, 2, 3

  // Remove consecutive duplicates
  print('distinct:');
  await Stream.fromIterable([1, 1, 2, 2, 3, 3]).distinct().forEach(print);  // 1, 2, 3
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

Operator	Function	Return Type
`where`	Filter elements	Stream`<T>`
`map`	Transform elements	Stream`<R>`
`expand`	One-to-many flattening	Stream`<T>`
`take`	Take first N elements	Stream`<T>`
`skip`	Skip first N elements	Stream`<T>`
`distinct`	Remove consecutive duplicates	Stream`<T>`
`takeWhile`	Take while condition holds	Stream`<T>`
`skipWhile`	Skip while condition holds	Stream`<T>`

6. StreamController and StreamSubscription

(1) StreamController

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：Complete StreamController Usage

DART

import 'dart:async';

class OrderStream {
  final _controller = StreamController<Map<String, dynamic>>();

  Stream<Map<String, dynamic>> get stream => _controller.stream;

  void addOrder(Map<String, dynamic> order) => _controller.add(order);

  void addError(Object error) => _controller.addError(error);

  void close() => _controller.close();
}

void main() async {
  final orderStream = OrderStream();

  // Subscribe to the stream
  final subscription = orderStream.stream.listen(
    (order) => print('Processing: ${order['id']}'),
    onError: (e) => print('Error: $e'),
    onDone: () => print('Stream closed'),
  );

  // Add orders
  orderStream.addOrder({'id': 'ORD-001', 'amount': 1500.0});
  orderStream.addOrder({'id': 'ORD-002', 'amount': 3200.0});
  orderStream.addOrder({'id': 'ORD-003', 'amount': 890.0});

  // Close the stream
  orderStream.close();

  // Wait for stream to finish
  await subscription.asFuture();
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

(2) Broadcast StreamController

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：Broadcast Mode

DART

import 'dart:async';

void main() async {
  // Broadcast controller - multiple listeners
  final controller = StreamController``<String>``.broadcast();

  // Multiple subscribers
  final sub1 = controller.stream.listen((e) => print('Sub1: $e'));
  final sub2 = controller.stream.listen((e) => print('Sub2: $e'));

  // Both subscribers receive the same event
  controller.add('ORD-001');
  controller.add('ORD-002');

  controller.close();
  await Future.delayed(const Duration(milliseconds: 100));
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

7. Bob's Scenario: Real-time Order Stream Processing

▶ Example

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

：Stream Pipeline Processing Orders

DART

import 'dart:async';

class Order {
  final String id;
  final double amount;
  final String status;
  final String category;

  Order({required this.id, required this.amount, required this.status, required this.category});

  @override
  String toString() => 'Order($id, \$${amount.toStringAsFixed(2)}, $status, $category)';
}

void main() async {
  // Simulate real-time order stream
  final controller = StreamController``<Order>``();

  // Build processing pipeline
  final revenueByCategory = <String, double>{};
  var totalOrders = 0;
  var completedOrders = 0;

  controller.stream
      .where((order) => order.amount > 0)       // Filter invalid
      .where((order) => order.status == 'completed')  // Only completed
      .map((order) => order)                     // Could transform here
      .listen(
        (order) {
          completedOrders++;
          totalOrders++;
          revenueByCategory.update(
            order.category,
            (v) => v + order.amount,
            ifAbsent: () => order.amount,
          );
          print('  Processed: ${order.id} - \$${order.amount.toStringAsFixed(2)} USD');
        },
        onDone: () {
          print('\n=== Stream Processing Complete ===');
          print('Completed orders: $completedOrders');
          for (final entry in revenueByCategory.entries) {
            print('  ${entry.key}: \$${entry.value.toStringAsFixed(2)} USD');
          }
          final total = revenueByCategory.values.fold(0.0, (a, b) => a + b);
          print('Total revenue: \$${total.toStringAsFixed(2)} USD');
        },
      );

  // Simulate incoming orders
  controller.add(Order(id: 'ORD-001', amount: 1500.0, status: 'completed', category: 'Electronics'));
  controller.add(Order(id: 'ORD-002', amount: -50.0, status: 'completed', category: 'Books'));      // Invalid
  controller.add(Order(id: 'ORD-003', amount: 3200.0, status: 'pending', category: 'Electronics'));  // Not completed
  controller.add(Order(id: 'ORD-004', amount: 890.0, status: 'completed', category: 'Clothing'));
  controller.add(Order(id: 'ORD-005', amount: 2100.0, status: 'completed', category: 'Electronics'));

  controller.close();
  await Future.delayed(const Duration(milliseconds: 100));
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

8. Complete Example: DataPipeline Stream Pipeline

DART

// ============================================
// DataPipeline Stream Processing Pipeline
// Real-time order stream with operators
// ============================================

import 'dart:async';

class Order {
  final String id;
  final double amount;
  final String status;
  final String category;
  final String region;

  const Order({
    required this.id,
    required this.amount,
    required this.status,
    required this.category,
    this.region = 'US',
  });

  double get taxAmount => amount * 0.08;
  double get totalWithTax => amount + taxAmount;

  @override
  String toString() =>
      'Order($id, \$${amount.toStringAsFixed(2)}, $status, $category, $region)';
}

class StreamPipeline {
  final _controller = StreamController``<Order>``();
  final Map<String, double> _revenue = {};
  final Map<String, int> _count = {};
  int _totalProcessed = 0;
  int _totalSkipped = 0;
  late StreamSubscription``<Order>`` _subscription;

  Stream``<Order>`` get stream => _controller.stream;

  StreamPipeline() {
    _subscription = _controller.stream
        .where((o) => o.amount > 0)
        .where((o) => o.status != 'cancelled')
        .distinct((o) => o.id)  // Deduplicate by ID
        .listen(
          _processOrder,
          onError: (e) => print('Pipeline error: $e'),
          onDone: _printSummary,
        );
  }

  void _processOrder(Order order) {
    _totalProcessed++;
    _revenue.update(
      order.category,
      (v) => v + order.totalWithTax,
      ifAbsent: () => order.totalWithTax,
    );
    _count.update(order.category, (v) => v + 1, ifAbsent: () => 1);
  }

  void addOrder(Order order) => _controller.add(order);

  void addError(Object error) => _controller.addError(error);

  Future``<void>`` close() async {
    await _controller.close();
    await _subscription.asFuture();
  }

  void _printSummary() {
    final totalRevenue = _revenue.values.fold(0.0, (a, b) => a + b);
    final totalCount = _count.values.fold(0, (a, b) => a + b);

    print('\n=== DataPipeline Stream Report ===');
    print('Processed: $_totalProcessed orders');
    print('Skipped:   $_totalSkipped records');
    print('Revenue:   \$${totalRevenue.toStringAsFixed(2)} USD');
    print('Orders:    $totalCount');

    print('\n--- By Category ---');
    final sorted = _revenue.entries.toList()
      ..sort((a, b) => b.value.compareTo(a.value));
    for (final entry in sorted) {
      final count = _count[entry.key] ?? 0;
      final avg = entry.value / count;
      print('  ${entry.key}:');
      print('    Orders:  $count');
      print('    Revenue: \$${entry.value.toStringAsFixed(2)} USD');
      print('    Average: \$${avg.toStringAsFixed(2)} USD');
    }
  }
}

void main() async {
  final pipeline = StreamPipeline();

  // Simulate real-time order stream
  final orders = [
    Order(id: 'ORD-001', amount: 1500.0, status: 'completed', category: 'Electronics'),
    Order(id: 'ORD-002', amount: -50.0, status: 'completed', category: 'Books'),
    Order(id: 'ORD-003', amount: 3200.0, status: 'pending', category: 'Electronics'),
    Order(id: 'ORD-004', amount: 890.0, status: 'completed', category: 'Clothing'),
    Order(id: 'ORD-001', amount: 1500.0, status: 'completed', category: 'Electronics'),  // Duplicate
    Order(id: 'ORD-005', amount: 2100.0, status: 'completed', category: 'Electronics'),
    Order(id: 'ORD-006', amount: 500.0, status: 'cancelled', category: 'Books'),
    Order(id: 'ORD-007', amount: 120.0, status: 'completed', category: 'Books'),
  ];

  // Emit orders one by one (simulating real-time)
  for (final order in orders) {
    pipeline.addOrder(order);
  }

  await pipeline.close();
}

TEXT 📖 Display only

> **Output:** Run locally in DartPad or with `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x. Results may vary slightly depending on the SDK version.

Output:

TEXT 📖 Display only

=== DataPipeline Stream Report ===
Processed: 5 orders
Skipped:   0 records
Revenue:   $8154.00 USD
Orders:    5

--- By Category ---
  Electronics:
    Orders:  3
    Revenue: $7236.00 USD
    Average: $2412.00 USD
  Clothing:
    Orders:  1
    Revenue: $961.20 USD
    Average: $961.20 USD
  Books:
    Orders:  1
    Revenue: $129.60 USD
    Average: $129.60 USD

❓ FAQ

Q: What is the difference between a Stream and a Future? A: A Future represents a single asynchronous result (0 or 1 value), while a Stream represents a sequence of asynchronous events (0 to many values). A Stream is a generalization of a Future — a multi-value version.

Q: Can a single-subscription Stream be listened to multiple times? A: No. A single-subscription Stream can only be listened to once; a second listen will throw a StateError. Use a broadcast Stream (via asBroadcastStream() or StreamController.broadcast()) when multiple listeners are needed.

Q: Are Stream operators lazy? A: Yes. Operators like map/where do not execute immediately; they only start processing when listened to. This is called a "cold stream".

Q: How do you handle errors in a Stream? A: There are three ways: the onError callback in listen, the handleError operator, or wrapping await for with try-catch. Using the onError callback in listen is recommended.

Q: Does a StreamController need to be closed? A: Yes. Call controller.close() to close the stream and notify listeners that the stream has ended. Not closing it will cause listeners to wait indefinitely.

Q: What are the limitations of a broadcast Stream? A: A broadcast Stream does not support pausing/resuming, and if a listener subscribes after an event has been emitted, it will miss previous events. It is suitable for "real-time subscription" scenarios.

Q: Does a Stream support backpressure? A: A single-subscription Stream naturally supports backpressure — when the listener (consumer) processes slowly, the producer will wait. A broadcast Stream does not support backpressure.

📖 Summary

A Stream is a sequence of asynchronous events, divided into single-subscription (1 listener) and broadcast (multiple listeners)
Creation methods: fromIterable (existing data), periodic (timed generation), StreamController (manual control)
Operator chain: where filter → map transform → expand flatten → take/skip limit → distinct deduplicate
StreamController is the production end of a Stream, while StreamSubscription is the consumption end
A Stream pipeline enables processing of millions of data pieces one by one, with stable memory usage and support for real-time processing

📝 Exercises

Basic (Difficulty ⭐): Use Stream.fromIterable to create a stream containing 10 amounts, use where to filter those greater than 500, use map to convert them to USD format, and use listen to print the results.
Intermediate (Difficulty ⭐⭐): Use StreamController to simulate a real-time order stream, adding 1 order per second for 5 seconds. Build a pipeline to filter completed orders, aggregate revenue by category, and finally output a report.
Challenge (Difficulty ⭐⭐⭐): Implement a "merge streams" function that combines two order streams into one (interleaved by time), deduplicates them, and then processes them. Hint: Use StreamGroup (from the async package) or manually merge with StreamController.

← Previous Lesson | Next Lesson →

Web-Tutorial Tech Team

A team of developers maintaining programming tutorials. Each tutorial is written and reviewed by developers with expertise in that field. We work to keep our content accurate and reliable — if you spot an issue, please let us know.

About Us

Dart: Dart Stream

1. What You Will Learn

2. A Developer's Real Story

(1) Pain Point: Millions of Data Records Cannot Be Loaded into Memory All at Once

(2) Stream Solution

(3) Benefits

3. Stream Basics

(1) Single-subscription vs Broadcast

▶ Example

4. Stream Creation

▶ Example

▶ Example

▶ Example

5. Stream Operators

▶ Example

▶ Example

▶ Example

▶ Example

6. StreamController and StreamSubscription

(1) StreamController

▶ Example

(2) Broadcast StreamController

▶ Example

7. Bob's Scenario: Real-time Order Stream Processing

▶ Example

8. Complete Example: DataPipeline Stream Pipeline

❓ FAQ

📖 Summary

📝 Exercises

🙏 帮我们做得更好