在 Flutter 裏使用 Stream

時間  2020-04-22
標籤 flutter 使用 stream 欄目 Apache 简体版
原文   原文鏈接

前言

在 Flutter 中有兩種處理異步操做的方式 Future 和 Stream，Future 用於處理單個異步操做，Stream 用來處理連續的異步操做。好比往水杯倒水，將一個水杯倒滿爲一個 Future，連續的將多個水杯倒滿就是 Stream。html

Stream 詳解

Stream 是一個抽象類，用於表示一序列異步數據的源。它是一種產生連續事件的方式，能夠生成數據事件或者錯誤事件，以及流結束時的完成事件。web

abstract class Stream<T> {
  Stream();
}
複製代碼

Stream 分單訂閱流和廣播流。編程

單訂閱流在發送完成事件以前只容許設置一個監聽器，而且只有在流上設置監聽器後纔開始產生事件，取消監聽器後將中止發送事件。即便取消了第一個監聽器，也不容許在單訂閱流上設置其餘的監聽器。廣播流則容許設置多個監聽器，也能夠在取消上一個監聽器後再次添加新的監聽器。緩存

Stream 有同步流和異步流之分。markdown

它們的區別在於同步流會在執行 add，addError 或 close 方法時當即向流的監聽器 StreamSubscription 發送事件，而異步流老是在事件隊列中的代碼執行完成後在發送事件。網絡

Stream 家族

StreamControllerapp

帶有控制流方法的流。 能夠向它的流發送數據，錯誤和完成事件，也能夠檢查數據流是否已暫停，是否有監聽器。sync 參數決定這個流是同步流仍是異步流。less

abstract class StreamController<T> implements StreamSink<T> {
  Stream<T> get stream;
  /// ...
}

StreamController _streamController = StreamController(
  onCancel: () {},
  onListen: () {},
  onPause: () {},
  onResume: () {},
  sync: false,
);
複製代碼

StreamSinkdom

流事件的入口。提供 add，addError，addStream 方法向流發送事件。異步

abstract class StreamSink<S> implements EventSink<S>, StreamConsumer<S> {
  Future close();
  /// ...
  Future get done;
}
複製代碼

StreamSubscription

流的監聽器。提供 cacenl、pause, resume 等方法管理。

abstract class StreamSubscription<T> {
  /// ...
}

StreamSubscription subscription = StreamController().stream.listen(print);
subscription.onDone(() => print('done'));
複製代碼

StreamBuilder

使用流數據渲染 UI 界面的部件。

StreamBuilder(
  // 數據流
  stream: stream,
  // 初始數據
  initialData: 'loading...',
  builder: (context, AsyncSnapshot snapshot) {
    // AsyncSnapshot 對象爲數據快照，緩存了當前數據和狀態
    // snapshot.connectionState
    // snapshot.data
    if (snapshot.hasData) {
      Map data = snapshot.data;
      return Text(data),
    }
    return CircularProgressIndicator();
  },
)
複製代碼

建立 Stream

在 Dart 有幾種方式建立 Stream

  1. 從現有的生成一個新的流 Stream，使用 map， where， takeWhile 等方法。 
// 整數流
Stream<int> intStream = StreamController<int>().stream;
// 偶數流
Stream<int> evenStream = intStream.where((int n) => n.isEven);
// 兩倍流
Stream<int> doubleStream = intStream.map((int n) => n * 2);
// 數字大於 10 的流
Stream<int> biggerStream = intStream.takeWhile((int n) => n > 10);
複製代碼
  1. 使用 async* 函數。 
Stream<int> countStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield i;
  }
}

Stream stream = countStream(10);
stream.listen(print);
複製代碼
  1. 使用 StreamController。 
StreamController<Map> _streamController = StreamController(
  onCancel: () {},
  onListen: () {},
  onPause: () {},
  onResume: () {},
  sync: false,
);

Stream _stream = _streamController.stream;
複製代碼
  1. 使用 Future 對象生成 
Future<int> _delay(int seconds) async {
  await Future.delayed(Duration(seconds: seconds));
  return seconds;
}

List<Future> futures = [];
for (int i = 0; i < 10; i++) {
  futures.add(_delay(3));
}

Stream _futuresStream = Stream.fromFutures(futures);
複製代碼

應用 Stream

Stream Counter

把 Flutter 的默認項目改用 Stream 實現

import 'dart:async';
import 'package:flutter/material.dart';

class StreamCounter extends StatefulWidget {
  @override
  _StreamCounterState createState() => _StreamCounterState();
}

class _StreamCounterState extends State<StreamCounter> {
  // 建立一個 StreamController
  StreamController<int> _counterStreamController = StreamController<int>(
    onCancel: () {
      print('cancel');
    },
    onListen: () {
      print('listen');
    },
  );

  int _counter = 0;
  Stream _counterStream;
  StreamSink _counterSink;

  // 使用 StreamSink 向 Stream 發送事件，當 _counter 大於 9 時調用 close 方法關閉流。
  void _incrementCounter() {
    if (_counter > 9) {
      _counterSink.close();
      return;
    }
    _counter++;
    _counterSink.add(_counter);
  }

  // 主動關閉流
  void _closeStream() {
    _counterStreamController.close();
  }

  @override
  void initState() {
    super.initState();
    _counterSink = _counterStreamController.sink;
    _counterStream = _counterStreamController.stream;
  }

  @override
  void dispose() {
    super.dispose();
    _counterSink.close();
    _counterStreamController.close();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('Stream Counter'),
      ),
      body: Center(
        child: Column(
          mainAxisAlignment: MainAxisAlignment.center,
          children: <Widget>[
            Text('You have pushed the button this many times:'),
            // 使用 StreamBuilder 顯示和更新 UI
            StreamBuilder<int>(
              stream: _counterStream,
              initialData: _counter,
              builder: (context, snapshot) {
                if (snapshot.connectionState == ConnectionState.done) {
                  return Text(
                    'Done',
                    style: Theme.of(context).textTheme.bodyText2,
                  );
                }

                int number = snapshot.data;
                return Text(
                  '$number',
                  style: Theme.of(context).textTheme.bodyText2,
                );
              },
            ),
          ],
        ),
      ),
      floatingActionButton: Row(
        mainAxisAlignment: MainAxisAlignment.center,
        children: <Widget>[
          FloatingActionButton(
            onPressed: _incrementCounter,
            tooltip: 'Increment',
            child: Icon(Icons.add),
          ),
          SizedBox(width: 24.0),
          FloatingActionButton(
            onPressed: _closeStream,
            tooltip: 'Close',
            child: Icon(Icons.close),
          ),
        ],
      ),
    );
  }
}

複製代碼

NetWork Status

監聽手機的網絡連接狀態，首先添加 connectivity 插件

dependencies:
  connectivity: ^0.4.8+2
複製代碼
import 'dart:async';
import 'package:connectivity/connectivity.dart';
import 'package:flutter/material.dart';

class NetWorkStatus extends StatefulWidget {
  @override
  _NetWorkStatusState createState() => _NetWorkStatusState();
}

class _NetWorkStatusState extends State<NetWorkStatus> {
  StreamController<ConnectivityResult> _streamController = StreamController();
  StreamSink _streamSink;
  Stream _stream;
  String _result;

  void _checkStatus() async {
    final ConnectivityResult result = await Connectivity().checkConnectivity();

    if (result == ConnectivityResult.mobile) {
      _result = 'mobile';
    } else if (result == ConnectivityResult.wifi) {
      _result = 'wifi';
    } else if (result == ConnectivityResult.none) {
      _result = 'none';
    }

    setState(() {});
  }

  @override
  void initState() {
    super.initState();
    _stream = _streamController.stream;
    _streamSink = _streamController.sink;
    _checkStatus();
    Connectivity().onConnectivityChanged.listen(
      (ConnectivityResult result) {
        _streamSink.add(result);
      },
    );
  }

  @override
  dispose() {
    super.dispose();
    _streamSink.close();
    _streamController.close();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('Network Status'),
      ),
      body: Center(
        child: StreamBuilder<ConnectivityResult>(
          stream: _stream,
          builder: (context, AsyncSnapshot snapshot) {
            if (snapshot.hasData) {
              if (snapshot.data == ConnectivityResult.mobile) {
                _result = 'mobile';
              } else if (snapshot.data == ConnectivityResult.wifi) {
                _result = 'wifi';
              } else if (snapshot.data == ConnectivityResult.none) {
                return Text('尚未連接網絡');
              }
            }

            if (_result == null) {
              return CircularProgressIndicator();
            }

            return ResultText(_result);
          },
        ),
      ),
    );
  }
}

class ResultText extends StatelessWidget {
  final String result;

  const ResultText(this.result);

  @override
  Widget build(BuildContext context) {
    return RichText(
      text: TextSpan(
        style: TextStyle(color: Colors.black),
        text: '正在使用',
        children: [
          TextSpan(
            text: ' $result ',
            style: TextStyle(
              color: Colors.red,
              fontSize: 20.0,
              fontWeight: FontWeight.bold,
            ),
          ),
          TextSpan(text: '連接網絡'),
        ],
      ),
    );
  }
}

複製代碼

Random Article

請求網絡數據建立流

dependencies:
  dio: ^3.0.9
  flutter_html: ^0.11.1
複製代碼
import 'dart:async';

import 'package:dio/dio.dart';
import 'package:flutter/material.dart';
import 'package:flutter_html/flutter_html.dart';

class RandomArticle extends StatefulWidget {
  @override
  _RandomArticleState createState() => _RandomArticleState();
}

class _RandomArticleState extends State<RandomArticle> {
  static Dio _dio = Dio(
    BaseOptions(baseUrl: 'https://interface.meiriyiwen.com'),
  );

  static Future<Map> _getArticle() async {
    Response response = await _dio.get(
      '/article/random',
      queryParameters: {"dev": 1},
    );

    final data = response.data['data'];
    return data;
  }

  Stream<Map> _futuresStream;

  @override
  void initState() {
    List<Future<Map>> futures = [];
    for (int i = 0; i < 10; i++) {
      // 添加 Future
      futures.add(_getArticle());
    }

    // 生成 Stream
    _futuresStream = Stream<Map>.fromFutures(futures);
    super.initState();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: Text('Random Article')),
      body: SingleChildScrollView(
        child: Center(
          child: StreamBuilder<Map>(
            stream: _futuresStream,
            builder: (context, AsyncSnapshot snapshot) {
              if (snapshot.hasData) {
                Map article = snapshot.data;

                return Container(
                  child: Column(
                    children: <Widget>[
                      SizedBox(height: 24.0),
                      Text(
                        article['title'],
                        style: TextStyle(fontSize: 24.0),
                      ),
                      Padding(
                        padding: const EdgeInsets.only(
                          top: 12.0,
                          left: 12.0,
                          right: 12.0,
                          bottom: 60.0,
                        ),
                        child: Html(
                          data: article['content'],
                        ),
                      ),
                    ],
                  ),
                );
              }
              return CircularProgressIndicator();
            },
          ),
        ),
      ),
    );
  }
}

複製代碼

Broadcast Stream

使用廣播流

import 'dart:async';

import 'package:flutter/material.dart';

class BroadcastStream extends StatefulWidget {
  @override
  _BroadcastStreamState createState() => _BroadcastStreamState();
}

class _BroadcastStreamState extends State<BroadcastStream> {
  StreamController<int> _streamController = StreamController<int>.broadcast();
  StreamSubscription _subscription1;
  StreamSubscription _subscription2;
  StreamSubscription _subscription3;

  int _count = 0;
  int _s1 = 0;
  int _s2 = 0;
  int _s3 = 0;

  @override
  void initState() {
    _subscription1 = _streamController.stream.listen((n) {
      setState(() {
        _s1 += 1;
      });
    });

    _subscription2 = _streamController.stream.listen((n) {
      setState(() {
        _s2 += 2;
      });
    });

    _subscription3 = _streamController.stream.listen((n) {
      setState(() {
        _s3 -= 1;
      });
    });

    super.initState();
  }

  void _add() {
    if (_count > 10) {
      // 大於 10 時中止第一個訂閱
      _subscription1.cancel();
    }
    _count++;
    _streamController.add(_count);
  }

  @override
  void dispose() {
    super.dispose();
    _streamController.close();
    _subscription1.cancel();
    _subscription2.cancel();
    _subscription3.cancel();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('Broadcast Stream'),
      ),
      body: Container(
        width: double.infinity,
        height: MediaQuery.of(context).size.height,
        child: Column(
          mainAxisAlignment: MainAxisAlignment.center,
          crossAxisAlignment: CrossAxisAlignment.center,
          children: [
            Text('Count: $_count'),
            SizedBox(height: 12.0),
            Text('S1: $_s1'),
            SizedBox(height: 12.0),
            Text('S2: $_s2'),
            SizedBox(height: 12.0),
            Text('S3: $_s3'),
            SizedBox(height: 12.0),
            FloatingActionButton(
              onPressed: _add,
              child: Icon(Icons.plus_one),
            ),
          ],
        ),
      ),
    );
  }
}

複製代碼

總結

Stream 是處理異步編程的方式之一，它提供一個了異步的事件序列，並在你準備好接受時發送。在 Dart 中流分爲同步流和異步流，以及單訂閱流和廣播流，有多種方式建立 Stream。

參考

異步編程：使用 stream

在 Dart 裏使用 Stream

全面深刻理解Stream

Building a Widget with StreamBuilder

StreamBuilder (Flutter 本週小部件)

Dart Streams - 聚焦 Flutter

本文使用 mdnice 排版

PS:【廣州】做者求職中歡迎聯繫。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程