DartのStream入門（非同期データの流れをつかむ）

Flutterで最初に詰まりやすいDartの書き方：final・const・null safety・async/await を最初に整理するの次に整理したいのが Stream です。Future が 1 回だけ返る値なら、Stream は時間経過とともに複数回流れてくる値です。この記事では、StreamController / listen / await for / StreamBuilder の 4 つに絞って、Flutter の画面でどこに出てくるかまで含めて確認します。

1. ゴールと非対象

対象読者

Flutter の環境構築と Dart 入門までは終わった人
Future は何となく読めるが、Stream が出ると処理の流れを追いにくい人
MethodChannel、接続状態、センサーデータ、StreamBuilder の前提を先に固めたい人

この記事で到達する状態

Future と Stream の違いを説明できる
StreamController から値を流し、listen で受ける最小例を読める
await for と listen を場面で使い分けられる
StreamBuilder で画面に値を流す入口を作れる
cancel / close / dispose をどこで呼ぶか判断できる

非対象

Riverpod / BLoC / RxDart
transform や where などの演算子の深掘り
broadcast と single-subscription の設計差の詳細
MethodChannel の実装そのもの
isolate

今回は Stream の入口だけに絞ります。状態管理やネイティブ連携の前に、まずは「複数回流れる値をどう読むか」を固める段階です。

2. まずは `Future` と `Stream` の違いを図で掴む

Future は「あとで 1 回返る値」です。Stream は「あとで 0 回以上流れてくる値」と考えると分かりやすくなります。最初はこの違いだけ押さえれば十分です。

型	返り方	まず合う場面	Flutter でよく見る例
`Future<T>`	最後に 1 回だけ返る	API 1 回呼び出し、保存処理、初回読込	`await http.get(...)`
`Stream<T>`	値が複数回流れる	監視、購読、継続更新	接続状態、センサー、バーコード検出、`StreamBuilder`

flowchart LR
  A[イベント発生] --> B[Stream に値が流れる]
  B --> C[listen / await for で受ける]
  C --> D[必要なら非同期処理]
  D --> E[StreamBuilder や setState で画面反映]

Future は「結果が返るまで待つ」読み方をします。Stream は「流れてくるたびに受ける」読み方です。MethodChannel のイベント受信、Bluetooth の接続状態、バーコード検出のように、値が 1 回で終わらない場面では Stream が自然に出てきます。

以降のコード例はすべて lib/main.dart にそのまま貼って flutter run で確認できます。外部パッケージ不要のため、DartPad でも試せます。

3. `StreamController` と `listen` で値の流れを確認する

最初は「流す側」と「受ける側」を同じ画面に置くと分かりやすくなります。StreamController が値の入口、listen が購読側です。

次の lib/main.dart は、値を流す操作と受信ログを同じ画面で見比べる最小例です。add、listen、cancel がどこにあるかを追うと流れが掴みやすくなります。

lib/main.dart は次の内容で作成します。

import 'dart:async';

import 'package:flutter/material.dart';

void main() {
  runApp(const MyApp());
}

class MyApp extends StatelessWidget {
  const MyApp({super.key});

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Stream listen sample',
      theme: ThemeData(
        colorScheme: ColorScheme.fromSeed(seedColor: Colors.teal),
      ),
      home: const StreamListenPage(),
    );
  }
}

class StreamListenPage extends StatefulWidget {
  const StreamListenPage({super.key});

  @override
  State<StreamListenPage> createState() => _StreamListenPageState();
}

class _StreamListenPageState extends State<StreamListenPage> {
  final StreamController<int> _controller = StreamController<int>();
  final List<String> _logs = <String>[];

  StreamSubscription<int>? _subscription;
  int _nextValue = 1;
  int? _latestValue;
  bool _isCanceled = false;

  @override
  void initState() {
    super.initState();
    _subscription = _controller.stream.listen((value) {
      setState(() {
        _latestValue = value;
        _logs.insert(0, 'listen が受信: $value');
      });
    });
  }

  @override
  void dispose() {
    _subscription?.cancel();
    _controller.close();
    super.dispose();
  }

  Future<void> _sendOne() async {
    if (_isCanceled) {
      return;
    }

    _controller.add(_nextValue);
    setState(() {
      _nextValue += 1;
    });
  }

  Future<void> _sendThree() async {
    if (_isCanceled) {
      return;
    }

    for (var i = 0; i < 3; i += 1) {
      _controller.add(_nextValue);
      setState(() {
        _nextValue += 1;
      });
      await Future<void>.delayed(const Duration(milliseconds: 400));
    }
  }

  Future<void> _cancelSubscription() async {
    await _subscription?.cancel();
    if (!mounted) {
      return;
    }

    setState(() {
      _subscription = null;
      _isCanceled = true;
      _logs.insert(0, '購読をキャンセルしました');
    });
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('StreamController + listen')),
      body: Padding(
        padding: const EdgeInsets.all(16),
        child: Column(
          crossAxisAlignment: CrossAxisAlignment.stretch,
          children: [
            Card(
              child: Padding(
                padding: const EdgeInsets.all(16),
                child: Column(
                  crossAxisAlignment: CrossAxisAlignment.start,
                  children: [
                    const Text(
                      '最新値',
                      style: TextStyle(
                        fontSize: 16,
                        fontWeight: FontWeight.bold,
                      ),
                    ),
                    const SizedBox(height: 8),
                    Text(
                      _latestValue == null ? '-' : _latestValue.toString(),
                      style: Theme.of(context).textTheme.headlineMedium,
                    ),
                    const SizedBox(height: 8),
                    Text(
                      _isCanceled ? '状態: キャンセル済み' : '状態: 購読中',
                    ),
                  ],
                ),
              ),
            ),
            const SizedBox(height: 16),
            Wrap(
              spacing: 12,
              runSpacing: 12,
              children: [
                ElevatedButton(
                  onPressed: _sendOne,
                  child: const Text('1件流す'),
                ),
                ElevatedButton(
                  onPressed: _sendThree,
                  child: const Text('3件流す'),
                ),
                OutlinedButton(
                  onPressed: _cancelSubscription,
                  child: const Text('購読を止める'),
                ),
              ],
            ),
            const SizedBox(height: 16),
            const Text(
              '受信ログ',
              style: TextStyle(fontSize: 16, fontWeight: FontWeight.bold),
            ),
            const SizedBox(height: 8),
            Expanded(
              child: ListView.separated(
                itemCount: _logs.length,
                separatorBuilder: (_, __) => const Divider(height: 1),
                itemBuilder: (context, index) {
                  return ListTile(
                    dense: true,
                    title: Text(_logs[index]),
                  );
                },
              ),
            ),
          ],
        ),
      ),
    );
  }
}

コードのポイント

① 値の入口は `StreamController<int>` が持つ

final StreamController<int> _controller = StreamController<int>();

この controller が、イベントを流し込む入口です。以降の _controller.add(...) と listen(...) は、どちらもこの stream を基準に動きます。

② 受信処理は `listen` に集約する

_subscription = _controller.stream.listen((value) {
  setState(() {
    _latestValue = value;
    _logs.insert(0, 'listen が受信: $value');
  });
});

値が届いたときに何をするかを callback にまとめているため、受信後の動きが1か所で読めます。画面更新とログ追加を同じ場所に置いているのも、入門例として追いやすい構成です。

③ 後始末は `cancel()` と `close()` をセットで行う

_subscription?.cancel();
_controller.close();

購読側と流す側の両方を閉じることで、不要なイベントやリソースを残さずに済みます。dispose() でまとめて後始末する流れは、Stream を扱う基本形です。

listen は UI と相性がよい形です。接続状態やスキャン結果のように、「流れてきたらすぐ画面やログへ反映したい」場面では callback で受けるほうが収まりよく読めます。

購読を止める を押したあとで 1件流す を押しても、画面は更新されません。値が流れていないのではなく、受け手を外した状態です。Stream を読むときは「いま値がないのか」「購読していないのか」を分けて考えると詰まりにくくなります。

4. `await for` と `listen` の違いを使い分ける

listen は流れてきた値を callback で受けます。await for は、ストリームの値を 1 件ずつ順番に取り出しながら、その途中で await を挟めます。

受け方	向く場面	読み方
`listen`	監視し続けたい、受信したら即反応したい	「流れてきたらこの callback を実行する」
`await for`	1件ずつ順番に非同期処理したい	「流れてくる間、ループで待ちながら処理する」

await for を使う最小例を lib/main.dart で確認します。このコードは、流れてきた注文を 1 件ずつ順番に処理する流れを見せるための例です。await for のループと、その中の待機処理を追います。

import 'dart:async';

import 'package:flutter/material.dart';

void main() {
  runApp(const MyApp());
}

class MyApp extends StatelessWidget {
  const MyApp({super.key});

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'await for sample',
      theme: ThemeData(
        colorScheme: ColorScheme.fromSeed(seedColor: Colors.orange),
      ),
      home: const AwaitForPage(),
    );
  }
}

class AwaitForPage extends StatefulWidget {
  const AwaitForPage({super.key});

  @override
  State<AwaitForPage> createState() => _AwaitForPageState();
}

class _AwaitForPageState extends State<AwaitForPage> {
  final List<String> _logs = <String>[];

  bool _isRunning = false;
  String _status = '未開始';

  Future<void> _startProcessing() async {
    if (_isRunning) {
      return;
    }

    setState(() {
      _isRunning = true;
      _status = '処理中';
      _logs.clear();
    });

    final Stream<String> orders = Stream<String>.periodic(
      const Duration(milliseconds: 600),
      (count) => 'ORDER-${count + 1}',
    ).take(3);

    await for (final order in orders) {
      if (!mounted) {
        return;
      }

      setState(() {
        _logs.add('受信: $order');
      });

      await Future<void>.delayed(const Duration(milliseconds: 500));

      if (!mounted) {
        return;
      }

      setState(() {
        _logs.add('登録完了: $order');
      });
    }

    if (!mounted) {
      return;
    }

    setState(() {
      _isRunning = false;
      _status = '完了';
    });
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('await for')),
      body: Padding(
        padding: const EdgeInsets.all(16),
        child: Column(
          crossAxisAlignment: CrossAxisAlignment.stretch,
          children: [
            Card(
              child: Padding(
                padding: const EdgeInsets.all(16),
                child: Column(
                  crossAxisAlignment: CrossAxisAlignment.start,
                  children: [
                    const Text(
                      '状態',
                      style: TextStyle(
                        fontSize: 16,
                        fontWeight: FontWeight.bold,
                      ),
                    ),
                    const SizedBox(height: 8),
                    Text(
                      _status,
                      style: Theme.of(context).textTheme.headlineMedium,
                    ),
                  ],
                ),
              ),
            ),
            const SizedBox(height: 16),
            ElevatedButton(
              onPressed: _isRunning ? null : _startProcessing,
              child: const Text('順番に処理する'),
            ),
            const SizedBox(height: 16),
            const Text(
              '処理ログ',
              style: TextStyle(fontSize: 16, fontWeight: FontWeight.bold),
            ),
            const SizedBox(height: 8),
            Expanded(
              child: ListView.separated(
                itemCount: _logs.length,
                separatorBuilder: (_, __) => const Divider(height: 1),
                itemBuilder: (context, index) {
                  return ListTile(
                    dense: true,
                    title: Text(_logs[index]),
                  );
                },
              ),
            ),
          ],
        ),
      ),
    );
  }
}

コードのポイント

① `await for` は流れてくる値を順番に取り出す

await for (final order in orders) {

listen の callback と違い、ここではループとして上から順に処理を書けます。受信した値ごとに非同期処理を挟みたいときに、流れが読みやすくなるのが利点です。

② 1 件ごとの処理の途中で `await` を挟める

await Future<void>.delayed(const Duration(milliseconds: 500));

各注文を受けてから登録完了まで少し待つ流れを、ループの中へそのまま書けます。ORDER-1 の処理が終わってから ORDER-2 へ進む、という順序が保たれる理由はここにあります。

③ 開始と完了の状態更新を前後で分ける

setState(() {
  _isRunning = true;
  _status = '処理中';
});

setState(() {
  _isRunning = false;
  _status = '完了';
});

処理の前後で状態を切り替えているため、画面側は「未開始」「処理中」「完了」を追えます。ストリーム処理そのものだけでなく、UI 上の状態遷移を読む練習にもなります。

この例では、ORDER-1 を受けたあとに 登録完了 の待ち時間を終えてから ORDER-2 へ進みます。await for は、こうした「各イベントごとに順番を守って非同期処理したい」場面に向いています。

listen が不要になるわけではありません。継続監視や UI の即時反応は listen が読みやすい場面が多く、1 件ずつ待ちながら処理したいなら await for のほうが収まりよく書けます。どちらが上ではなく、処理の形が違います。

5. `StreamBuilder` で画面に流す

listen で setState を呼ぶ方法でも画面更新はできます。とはいえ、ストリームをそのまま UI に渡したい場面では StreamBuilder のほうが読みやすくなります。

次の lib/main.dart は、進捗値を stream で流し、そのままカード UI に描画する例です。stream、snapshot.data、connectionState の3つの見方を確認します。

lib/main.dart を次の内容で作成します。

import 'dart:async';

import 'package:flutter/material.dart';

void main() {
  runApp(const MyApp());
}

class MyApp extends StatelessWidget {
  const MyApp({super.key});

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'StreamBuilder sample',
      theme: ThemeData(
        colorScheme: ColorScheme.fromSeed(seedColor: Colors.indigo),
      ),
      home: const StreamBuilderPage(),
    );
  }
}

class StreamBuilderPage extends StatefulWidget {
  const StreamBuilderPage({super.key});

  @override
  State<StreamBuilderPage> createState() => _StreamBuilderPageState();
}

class _StreamBuilderPageState extends State<StreamBuilderPage> {
  late Stream<int> _progressStream;

  @override
  void initState() {
    super.initState();
    _progressStream = _createProgressStream();
  }

  Stream<int> _createProgressStream() {
    return Stream<int>.periodic(
      const Duration(milliseconds: 500),
      (count) => (count + 1) * 20,
    ).take(5);
  }

  void _restart() {
    setState(() {
      _progressStream = _createProgressStream();
    });
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('StreamBuilder')),
      body: Padding(
        padding: const EdgeInsets.all(16),
        child: Column(
          crossAxisAlignment: CrossAxisAlignment.stretch,
          children: [
            StreamBuilder<int>(
              stream: _progressStream,
              initialData: 0,
              builder: (context, snapshot) {
                final int progress = snapshot.data ?? 0;
                final bool isDone = snapshot.connectionState == ConnectionState.done;

                return Card(
                  child: Padding(
                    padding: const EdgeInsets.all(16),
                    child: Column(
                      crossAxisAlignment: CrossAxisAlignment.start,
                      children: [
                        const Text(
                          '非同期進捗',
                          style: TextStyle(
                            fontSize: 16,
                            fontWeight: FontWeight.bold,
                          ),
                        ),
                        const SizedBox(height: 12),
                        LinearProgressIndicator(value: progress / 100),
                        const SizedBox(height: 12),
                        Text('進捗: $progress%'),
                        const SizedBox(height: 4),
                        Text(isDone ? '状態: 完了' : '状態: 進行中'),
                      ],
                    ),
                  ),
                );
              },
            ),
            const SizedBox(height: 16),
            ElevatedButton(
              onPressed: _restart,
              child: const Text('もう一度流す'),
            ),
          ],
        ),
      ),
    );
  }
}

コードのポイント

① `stream:` に渡した値が流れるたびに `builder` が呼ばれる

StreamBuilder<int>(
  stream: _progressStream,
  initialData: 0,

_progressStream が新しい値を流すたびに、builder が最新の snapshot を受け取ります。自前で購読と解除を書かずに、UI 側へストリームを直接つなげられるのが StreamBuilder の利点です。

② 最新値は `snapshot.data` から読む

final int progress = snapshot.data ?? 0;

ストリームがまだ値を流していない初期状態でも描画できるよう、initialData と ?? 0 を組み合わせています。画面に出したい値の入口がこの1行にまとまっています。

③ 進行中か完了かは `connectionState` で分ける

final bool isDone = snapshot.connectionState == ConnectionState.done;

値そのものとは別に、ストリームの状態も UI へ反映できます。進捗表示や接続状態のように「最新値」と「今どの段階か」を同時に見せたい場面で有効です。

FutureBuilder は 1 回の結果を描く場面に向きます。StreamBuilder は進捗、接続状態、継続監視のように値が何度も変わる場面で使います。Future と Stream の違いが、ここでもそのまま表れます。

6. キャンセルとリソース解放を忘れない

Stream を使い始めた段階で省きたくないのが後始末です。画面が閉じられても購読やコントローラーが残ると、不要な更新や例外の原因になります。

listen を使ったら StreamSubscription を保持し、不要になった時点で cancel() する
StreamController を自分で作ったら close() する
await のあとで setState する可能性があるなら mounted を確認する
画面の寿命に合わせるなら dispose() にまとめる

Stream の後始末は、次に出てくる FlutterのWidgetライフサイクル入門（initState / dispose で詰まらないために）ともつながります。initState で購読開始、dispose で終了、という流れをここで一度見ておくと、Controller や Animation の記事でも読み方が揃います。

7. まとめ

Stream 入門で最初に押さえたいのは次の 3 つです。

値が複数回流れるなら Stream
監視して反応するなら listen、順番に待ちながら処理するなら await for
画面へ流す入口は StreamBuilder

ここまで読めるようになると、Stream が出てきたときに「何が流れていて、どこで受けているか」を追いやすくなります。次に UI 基礎へ進むなら Flutterのレイアウト入門（Column / Row / Stack の使い分け）、非同期通信の実例へ進むなら FlutterからREST APIを呼ぶ最小構成（JSON通信 + エラー処理）を続けて読むと流れがつながります。

DartのStream入門（非同期データの流れをつかむ）

この記事の前提

目次

1. ゴールと非対象

対象読者

この記事で到達する状態

非対象

2. まずは `Future` と `Stream` の違いを図で掴む

3. `StreamController` と `listen` で値の流れを確認する

コードのポイント

① 値の入口は `StreamController<int>` が持つ

② 受信処理は `listen` に集約する

③ 後始末は `cancel()` と `close()` をセットで行う

4. `await for` と `listen` の違いを使い分ける

コードのポイント

① `await for` は流れてくる値を順番に取り出す

② 1 件ごとの処理の途中で `await` を挟める

③ 開始と完了の状態更新を前後で分ける

5. `StreamBuilder` で画面に流す

コードのポイント

① `stream:` に渡した値が流れるたびに `builder` が呼ばれる

② 最新値は `snapshot.data` から読む

③ 進行中か完了かは `connectionState` で分ける

6. キャンセルとリソース解放を忘れない

7. まとめ

このシリーズ

この記事の前提

目次

1. ゴールと非対象

対象読者

この記事で到達する状態

非対象

2. まずは Future と Stream の違いを図で掴む

3. StreamController と listen で値の流れを確認する

コードのポイント

① 値の入口は StreamController<int> が持つ

② 受信処理は listen に集約する

③ 後始末は cancel() と close() をセットで行う

4. await for と listen の違いを使い分ける

コードのポイント

① await for は流れてくる値を順番に取り出す

② 1 件ごとの処理の途中で await を挟める

③ 開始と完了の状態更新を前後で分ける

5. StreamBuilder で画面に流す

コードのポイント

① stream: に渡した値が流れるたびに builder が呼ばれる

② 最新値は snapshot.data から読む

③ 進行中か完了かは connectionState で分ける

6. キャンセルとリソース解放を忘れない

7. まとめ

このシリーズ

関連記事

FlutterのTabBarとBottomNavigationBarで複数画面を切り替える

Flutterのダイアログ・スナックバー・ボトムシートを使う（確認・通知UIの基本）

FlutterのListViewとGridViewで一覧画面を作る（基本パターン）

2. まずは `Future` と `Stream` の違いを図で掴む

3. `StreamController` と `listen` で値の流れを確認する

① 値の入口は `StreamController<int>` が持つ

② 受信処理は `listen` に集約する

③ 後始末は `cancel()` と `close()` をセットで行う

4. `await for` と `listen` の違いを使い分ける

① `await for` は流れてくる値を順番に取り出す

② 1 件ごとの処理の途中で `await` を挟める

5. `StreamBuilder` で画面に流す

① `stream:` に渡した値が流れるたびに `builder` が呼ばれる

② 最新値は `snapshot.data` から読む

③ 進行中か完了かは `connectionState` で分ける