EventBus介绍

在Android事件总线库中，EventBus库是最普遍之一。看看Android EventBus库的说明官网git

发布者/订阅者模式
集中式通讯
低耦合，简化组件通讯

自定义EventBus

咱们要实现一个Flutter版的EventBus！功能以下：github

注册Subscriber到EventBus
Publisher post Event，分发给Subscriber
EventBus能够注册多种类型的Event

Demo地址入口 main_eventbus.dart编程

接口定义

//发布者接口
abstract class IPublisher {
  void post<T>(T event);
}
//订阅者:函数对象
typedef ISubscriber<T> = void Function(T event);

//集中式通讯，
//1.IEventBus继承IPublisher,分发数据
//2.IEventBus注册和取消注册ISubscriber
abstract class IEventBus extends IPublisher {
  
  void register<T>(ISubscriber<T> subscriber);

  void unregister<T>(ISubscriber<T> subscriber);
}
复制代码

接口实现1

Type typeOf<T>() => T;

class MyEventBus1 implements IEventBus {
  //咱们用map存放咱们的订阅者。不一样订阅者订阅的Event类型多是不一样的
  Map<Type, List<Function>> map = new Map();

  @override
  void register<T>(ISubscriber<T> subscriber) {
    Type type = typeOf<T>();
    if (!map.containsKey(type)) {
      map[type] = new List();
    }
    map[type].add(subscriber);
  }

  @override
  void unregister<T>(ISubscriber<T> subscriber) {
    Type type = typeOf<T>();
    if (map.containsKey(type)) {
      map[type].remove(subscriber);
    }
  }

  //发布
  @override
  void post<T>(T event) {
    Type type = typeOf<T>();
    if (map.containsKey(type)) {
      var subscribers = map[type];
      subscribers?.forEach((subscriber) => subscriber?.call(event));
    }
  }
}
复制代码

接口实现2

class MyEventBus2 implements IEventBus {
  List<Function> subscribers = new List();

  @override
  register<T>(ISubscriber<T> subscriber) {
    if (!subscribers.contains(subscriber)) {
      subscribers.add(subscriber);
    }
  }

  @override
  unregister<T>(ISubscriber<T> subscriber) {
    if (subscribers.contains(subscriber)) {
      subscribers.remove(subscriber);
    }
  }

  @override
  post<T>(T event) {
    var ints = subscribers.whereType<ISubscriber<T>>();
    ints?.forEach((subscriber) => subscriber?.call(event));
  }
}
复制代码

功能测试

class EventX {}

class EventY {}

main() {
  testEventBus(new MyEventBus1());
  print("--------------------");
  testEventBus(new MyEventBus2());
}

void testEventBus(IEventBus eventBus) {
  ISubscriber<EventX> subscriber1 = (event) => print(event.toString());
  ISubscriber<EventX> subscriber2 = (event) => print(event.toString());
  eventBus.register(subscriber1);
  eventBus.register(subscriber2);
  eventBus.unregister(subscriber1);

  ISubscriber<EventY> subscriber3 = (event) => print(event.toString());
  ISubscriber<EventY> subscriber4 = (event) => print(event.toString());
  eventBus.register(subscriber3);
  eventBus.register(subscriber4);

  eventBus.post(new EventX());
  eventBus.post(new EventY());
}
复制代码

结果输出为如下，测试经过redux

Instance of 'EventX'
Instance of 'EventY'
Instance of 'EventY'
--------------------
Instance of 'EventX'
Instance of 'EventY'
Instance of 'EventY'
复制代码

使用自定义EventBus

需求效果图

点击每行的按钮，每行上的两个数字会增加1 markdown

代码实现

MyEventBus1 _eventBus = new MyEventBus1();
//首先咱们定义event。第一排用EventA，第二排用EventB
class BaseEvent { int count = 0;}
class EventA extends BaseEvent {}
class EventB extends BaseEvent {}

//点击按钮
class ButtonWidget<T extends BaseEvent> extends StatelessWidget {
  final T event;

  const ButtonWidget({Key key, this.event}) : super(key: key);

  @override
  Widget build(BuildContext context) {
    return RaisedButton(
      child: Text("increment"),
      onPressed: _increment,
    );
  }

  //点击处理
  void _increment() {
    if (event != null) {
      event.count++;
      _eventBus.post(event);
    }
  }
}

//显示数字
class TextWidget<T> extends StatefulWidget {
  @override
  _TextWidgetState<T> createState() {
    return _TextWidgetState<T>();
  }
}

class _TextWidgetState<T> extends State<TextWidget<T>> {
  int _count = 0;
  ISubscriber<T> _subscriber;

  @override
  void initState() {
    super.initState();
    //经过setState 刷新UI。
    
    _subscriber =
        (event) => setState(() => _count = (event as BaseEvent).count);
   //注册
    _eventBus.register<T>(_subscriber);
  }

  @override
  Widget build(BuildContext context) {
    return Text(
      " $_count ",
      style: TextStyle(fontSize: 18),
    );
  }

  @override
  void dispose() {
    super.dispose();
    //取消注册
    _eventBus.unregister<T>(_subscriber);
  }
}

//页面主体
class EventBusDemoWidget1 extends StatefulWidget {
  EventBusDemoWidget1({Key key}) : super(key: key);

  @override
  _EventBusDemoWidget1State createState() => _EventBusDemoWidget1State();
}

class _EventBusDemoWidget1State extends State<EventBusDemoWidget1> {
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text(eventBusDemo1Title),
      ),
      body: Container(
        child: Column(
          children: <Widget>[
            Row(
              children: <Widget>[
                new ButtonWidget(event: new EventA()),
                new TextWidget<EventA>(),
                new TextWidget<EventA>()
              ],
            ),
            Divider(
              color: Colors.grey,
            ),
            Row(
              children: <Widget>[
                new ButtonWidget(event: new EventB()),
                new TextWidget<EventB>(),
                new TextWidget<EventB>()
              ],
            )
          ],
        ),
      ),
    );
  }
}
复制代码

自定义EventBus大功告成！再来看看别人家的EventBus。app

pub上的EventBus

咱们在pub.dev上找到了一个EventBus库，github地址less

An Event Bus using Dart Streams for decoupling applications异步

使用EventBus

实际相对于自定义的EventBus,不一样点在ButtonWidget和TextWidget中。async

class ButtonWidget<T extends BaseEvent> extends StatelessWidget {
  final T event;

  const ButtonWidget({Key key, this.event}) : super(key: key);

  @override
  Widget build(BuildContext context) {
    return RaisedButton(
      child: Text("increment"),
      onPressed: _increment,
    );
  }

  void _increment() {
    if (event != null) {
      event.count++;
      print(event.count);
      _eventBus.fire(event);
// _eventBus.post(event);
    }
  }
}

class TextWidget<T> extends StatefulWidget {
  @override
  _TextWidgetState<T> createState() {
    return _TextWidgetState<T>();
  }
}

class _TextWidgetState<T> extends State<TextWidget<T>> {
  int _count = 0;
// ISubscriber<T> _subscriber;
  StreamSubscription<T> _subscriber;

  @override
  void initState() {
    super.initState();
// _subscriber =
// (event) => setState(() => _count = (event as BaseEvent).count);
// _eventBus.register<T>(_subscriber);
    _subscriber = _eventBus
        .on<T>()
        .listen((event) => setState(() => _count = (event as BaseEvent).count));
  }

  @override
  Widget build(BuildContext context) {
    print(typeOf<T>());
    return Text(
      " $_count ",
      style: TextStyle(fontSize: 18),
    );
  }

  @override
  void dispose() {
    super.dispose();
    // _eventBus.unregister<T>(_subscriber);
    _subscriber.cancel();
  }
}
  
复制代码

EventBus库源码

class EventBus {
  StreamController _streamController;

  StreamController get streamController => _streamController;

  EventBus({bool sync = false})
      : _streamController = StreamController.broadcast(sync: sync);

  EventBus.customController(StreamController controller)
      : _streamController = controller;

  Stream<T> on<T>() {
    if (T == dynamic) {
      return streamController.stream;
    } else {
      return streamController.stream.where((event) => event is T).cast<T>();
    }
  }

  void fire(event) {
    streamController.add(event);
  }

  void destroy() {
    _streamController.close();
  }
}
复制代码

EventBus库之Stream

不由感慨，Dart自带Stream，处理异步事件。Stream是Dart:async库的核心API，对异步提供了很是好的支持。这里简单贴图，体会下Stream。 ide

值得注意的是这句话

return streamController.stream.where((event) => event is T).cast<T>();
复制代码

返回的Stream包了两层 CastStream(_WhereStream(_BroadcastStream)，经过这样的处理就能够监听泛型事件了。

_WhereStream先处理

class _WhereStream<T> extends _ForwardingStream<T, T> {
  final _Predicate<T> _test;

  _WhereStream(Stream<T> source, bool test(T value))
      : _test = test,
        super(source);

    //_test即为 (event) => event is T
  void _handleData(T inputEvent, _EventSink<T> sink) {
    bool  satisfies = _test(inputEvent);
    if (satisfies) {
      sink._add(inputEvent);
    }
  }
}
复制代码

CastStreamSubscription实际作的类型转换

class CastStreamSubscription<S, T> implements StreamSubscription<T> {
  final StreamSubscription<S> _source;

  /// User's data handler. May be null.
  void Function(T) _handleData;

  CastStreamSubscription(this._source) {
    _source.onData(_onData);
  }
  void _onData(S data) {
    if (_handleData == null) return;
    T targetData = data as T;
    _zone.runUnaryGuarded(_handleData, targetData);
  }
复制代码

后记

至此，咱们在Provider库StreamProvider中遇到了Stream，咱们在EventBus库中遇到了Stream Dart语言自带优秀的异步编程模型 Stream和Future。 Stream拥有良好的异步模型，滋养了一大波组件库 Provider/BLoC/flutter_redux/Rxdart/fish-redux。接下来就有必要对Stream进行分析了。

Flutter中的事件总线(EventBus)