Flutter状态管理：ReactiveX之RxDart

一、前言

此处系列章节目录，待更新

• BLoC《Flutter状态管理：BLoC（Business Logic Component）》
• 编程概念《编程方式：命令式 vs 函数式》

二、StreamController增加版：Subject

其实无论从订阅或者变换都可以看出， Dart 中的 Stream 已经自带了类似 rx 的效果，但是为了让 rx 的用户们更方便的使用，ReactiveX 就封装了 rxdart 来满足用户的熟悉感，如下图所示为它们的对应关系：

Dart	RxDart
StreamController	Subject
Stream	Observable

在 rxdart 中， Observable 是一个 Stream，而 Subject 继承了 Observable 也是一个 Stream，并且 Subject 实现了 StreamController 的接口，所以它也具有 Controller 的作用。

如下代码所示是 rxdart 的简单使用，可以看出它屏蔽了外界需要对 StreamSubscription 和 StreamSink 等的认知，更符合 rx 历史用户的理解

final subject = PublishSubject<String>();

subject.stream.listen(observerA);
subject.add("AAAA1");
subject.add("AAAA2"));

subject.stream.listen(observeB);
subject.add("BBBB1");
subject.close();

以上方代码为例：

• PublishSubject 内部实际创建是创建了一个广播 StreamController<T>.broadcast ；
• 当我们调用 add 或者 addStream 时，最终会调用到的还是我们创建的 StreamController.add；
• 当我们调用 onListen 时，也是将回调设置到 StreamController 中。
• rxdart 在做变换时，我们获取到的 Observable 就是 this，也就是 PublishSubject 自身这个 Stream ，而 Observable 一系列的变换，也是基于创建时传入的 stream 对象，比如：

  @override
  Observable<S> asyncMap<S>(FutureOr<S> convert(T value)) =>Observable<S>(_stream.asyncMap(convert));

所以我们可以看出来，rxdart 只是对 Stream 进行了概念变换，变成了我们熟悉的对象和操作符，而这也是为什么 rxdart 可以在 StreamBuilder 中直接使用的原因。

RxDart提供了三种StreamController的变体来应用到不同的场景：

• PublishSubject
• BehaviorSubject
• ReplaySubject

以下来分别讲序这三种场景的使用情况。

2.1、PublishSubject

PublishSubject.png

PublishSubject最常见，从图中可看到，listener只能监听到订阅之后的事件：

final subject = PublishSubject();

subject.stream.listen((event) => print("observer1 => $event"));
subject.add(1);
subject.add(2);

subject.stream.listen((event) => print("observer2 => $event"));
subject.add(3);
subject.close();

// 打印输出：
// flutter: observer1 => 1
// flutter: observer2 => 3
// flutter: observer1 => 2
// flutter: observer1 => 3

2.2、BehaviorSubject

BehaviorSubject.png

BehaviorSubject也是广播，与PublishSubject的区别是：它会返回订阅前的最后一次事件：

final subject = BehaviorSubject();

subject.stream.listen((event) => print("observer1 => $event"));
subject.add(1);
subject.add(2);

subject.stream.listen((event) => print("observer2 => $event"));
subject.add(3);
subject.close();

// 打印输出：
// flutter: observer1 => 1
// flutter: observer2 => 2
// flutter: observer2 => 3
// flutter: observer1 => 2
// flutter: observer1 => 3

2.3、ReplaySubject

顾名思义：回放！会将订阅前的事件都发送给新的订阅者：

final subject = ReplaySubject();

subject.stream.listen((event) => print("observer1 => $event"));
subject.add(1);
subject.add(2);

subject.stream.listen((event) => print("observer2 => $event"));
subject.add(3);
subject.close();

// 打印输出：
// flutter: observer1 => 1
// flutter: observer2 => 1
// flutter: observer2 => 2
// flutter: observer2 => 3
// flutter: observer1 => 2
// flutter: observer1 => 3

三、实战演练

3.1、新建Model（CountModel.dart）

import 'package:rxdart/rxdart.dart';

class CountModel {
  BehaviorSubject _subject = BehaviorSubject.seeded(0);

  get stream => _subject.stream;
  get value => _subject.value;

  increment() {
    _subject.add(value + 1);
  }

  decrement() {
    _subject.add(value - 1);
  }
}

3.2、新建页面（RxdartPage.dart）

import 'package:flutter/material.dart';
import 'package:stateresearch/model/CountModel.dart';

class RxdartPage extends StatelessWidget {
  final CountModel _model = CountModel();

  @override
  Widget build(BuildContext context) {
    return _body(context);
  }

  Widget _body(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text("RxdartPage"),
      ),
      body: Center(
        child: Column(
          mainAxisAlignment: MainAxisAlignment.center,
          children: <Widget>[
            Text('You have pushed the button this many times:'),
            StreamBuilder(
              stream: _model.stream,
              builder: (BuildContext context, AsyncSnapshot snapshot) {
                return Text("${snapshot.data}");
              },
            ),
          ],
        ),
      ),
      floatingActionButton: FloatingActionButton(
        onPressed: _model.increment,
        tooltip: 'Increment',
        child: Icon(Icons.add),
      ),
    );
  }
}

3.3、修改main文件

import 'package:flutter/material.dart';
import 'package:stateresearch/pages/RxdartPage.dart';

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

class MyApp extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Flutter状态管理',
      theme: ThemeData(
        primarySwatch: Colors.blue,
        visualDensity: VisualDensity.adaptivePlatformDensity,
      ),
      home: RxdartPage(),
    );
  }
}

以上就是简单的局部状态管理的例子，至于全局共享，与BLoC类似，建个BLoC和Provider，再包裹MyApp就行。