Flutter 从源码扒一扒Stream机制

Stream的基本使用

//1、创建一个流控制对象，只要用来控制流的暂停、取消和订阅
StreamController _controller = StreamController();//2、实现对一个流的订阅和监听事件
_controller.stream.listen((event) {
print("event==$event");
});//3、添加一个事件
_controller.add("123");

StreamController类

职责是一个抽象类，用于创建一个可以发送数据和接收数据的可监听对象。
_StreamController 是StreamController的真正实现类
工厂构造方法

factory StreamController({void onListen()?,void onPause()?,void onResume()?,FutureOr<void> onCancel()?,bool sync = false}) {return sync? _SyncStreamController<T>(onListen, onPause, onResume, onCancel): _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);

我们使用StreamController的时候会这样
StreamController controller = StreamController();
但是StreamController是个抽象，不能被实例化，主要他使用了工厂构造，最后使用他的子类去实例化。
还要一个字段 bool sync = false ，false表示的是异步的，true表示同步。不更改的话，默认是异步的

_SyncStreamController和_AsyncStreamController就是StreamController的子类

//异步控制器
class _AsyncStreamController<T> = _StreamController<T>with _AsyncStreamControllerDispatch<T>;//同步控制器
class _SyncStreamController<T> = _StreamController<T>with _SyncStreamControllerDispatch<T>;

可以看到_AsyncStreamController和_SyncStreamController直接继承了_StreamController。

而_StreamController是StreamController的真正的实现类

//继承关系链
abstract class _StreamController<T> implements _StreamControllerBase<T> abstract class _StreamControllerBase<T>implementsStreamController<T>,_StreamControllerLifecycle<T>,_EventSink<T>,_EventDispatch<T> {

_controller.stream就是Stream对象

Stream 提供了一种接收事件序列的方法。每个事件要么是数据事件（也称为流的元素），要么是错误事件（这是某件事情失败的通知）。

这个流机制也算是一个生产者和消费者模式。
生产者就是_controller.add(“123”); 消费者就是listen监听了
那我们来看看stream内部是如何实现这个机制的?

listen监听

我们得先注册一下监听事件，添加对此流的订阅，才能接收到生产者的通知。

  StreamSubscription<T> listen(void onData(T event)?,{Function? onError, void onDone()?, bool? cancelOnError});StreamSubscription<T> listen(void onData(T data)?,{Function? onError, void onDone()?, bool? cancelOnError}) {cancelOnError ??= false;StreamSubscription<T> subscription =_createSubscription(onData, onError, onDone, cancelOnError);_onListen(subscription);return subscription;}这个listen方法是Stream类的方法，返回的是一个订阅对象StreamSubscription。并且是空的实现，需要子类去实现它。我们看看是哪个类实现了Stream类。它有很多的实现类，我选一个主要的类_StreamImpl。_StreamImpl 是 Stream的继承类，它也是个抽象类，我去找它的最终实现类_ControllerStreamabstract class _StreamImpl<T> extends Stream<T>class _ControllerStream<T> extends _StreamImpl<T>我们可以发现在_StreamController 有个方法，用来获取流对象的，正好是Stream的子类_ControllerStreamStream<T> get stream => _ControllerStream<T>(this);我们来看看_ControllerStream类StreamSubscription<T> _createSubscription(void onData(T data)?,Function? onError, void onDone()?, bool cancelOnError) =>_controller._subscribe(onData, onError, onDone, cancelOnError);

_controller就是抽象类_StreamControllerLifecycle的实例对象，
_subscribe是个抽象方法，需要子类去实现它。

abstract class _StreamController<T> implements _StreamControllerBase<T>
abstract class _StreamControllerBase<T>implementsStreamController<T>,_StreamControllerLifecycle<T>,EventSink<T>,_EventDispatch<T>

从上面的代码可以看出，最终是_StreamController实现了_subscribe方法
我们把_subscribe方法代码拿出来

StreamSubscription<T> _subscribe(void onData(T data)?, Function? onError,void onDone()?, bool cancelOnError) {if (!_isInitialState) {throw StateError("Stream has already been listened to.");}//创建一个订阅对象_ControllerSubscription<T> subscription = _ControllerSubscription<T>(this, onData, onError, onDone, cancelOnError);_PendingEvents<T>? pendingEvents = _pendingEvents;_state |= _STATE_SUBSCRIBED;if (_isAddingStream) {_StreamControllerAddStreamState<T> addState = _varData as dynamic;addState.varData = subscription;addState.resume();} else {_varData = subscription;}subscription._setPendingEvents(pendingEvents);subscription._guardCallback(() {_runGuarded(onListen);});return subscription;}

这个方法是关键，从方法名字我们就可以知道这是一个订阅方法，实现对一个事件的订阅。
我们看一下_ControllerSubscription，它的父类是StreamSubscription。

StreamSubscription 类

StreamSubscription 是一个抽象接口类
文档上描述订阅向listen提供事件，并保存用于处理事件的回调。订阅还可用于取消订阅事件，或暂时暂停流中的事件。

下面是它的一些抽象方法

    ///取消订阅事件Future<void> cancel();///处理订阅事件void onData(void handleData(T data)?);///完成订阅事件void onDone(void handleDone()?);///暂停订阅事件void pause([Future<void>? resumeSignal]);///恢复订阅事件void resume();

class _BufferingStreamSubscription<T>implements StreamSubscription<T>, _EventSink<T>, _EventDispatch<T>class _ControllerSubscription<T> extends _BufferingStreamSubscription<T>

发起数据通知

接着我们再看看数据添加方法，这个方法就是用来发起数据通知的

   if (hasListener) {_sendData(value);} else if (_isInitialState) {_ensurePendingEvents().add(_DelayedData<T>(value));}}// 这个_sendData方法是抽象类_EventDispatch的方法abstract class _EventDispatch<T> {void _sendData(T data);void _sendError(Object error, StackTrace stackTrace);void _sendDone();}

经过查找发现 _AsyncStreamControllerDispatch类实现了_sendData的方法

继承关系链

abstract class _StreamController<T> implements _StreamControllerBase<T>
abstract class _StreamControllerBase<T> implements StreamController<T>, _StreamControllerLifecycle<T>,_EventSink<T>,_EventDispatch<T> {}_AsyncStreamControllerDispatch类实现的方法如下void _sendData(T data) {_subscription._addPending(_DelayedData<T>(data));}_subscription 就是_ControllerSubscription的实例对象
如下代码所示_ControllerSubscription<T> get _subscription {assert(hasListener);Object? varData = _varData;if (_isAddingStream) {_StreamControllerAddStreamState<Object?> streamState = varData as dynamic;varData = streamState.varData;}return varData as dynamic;}varData 这个很熟悉就是上面的_subscribe方法里面实例化ControllerSubscription对象赋值给varData，这个varData实际就是ControllerSubscription对象。

继承关系链

class _ControllerSubscription<T> extends _BufferingStreamSubscription<T>
class _BufferingStreamSubscription<T> implements StreamSubscription<T>, _EventSink<T>, _EventDispatch<T>

接着我们再来看看 _subscription._addPending(_DelayedData(data));

代码如下，这个方法是_BufferingStreamSubscription类实现的

   void _addPending(_DelayedEvent event) {var pending = _pending ??= _PendingEvents<T>();pending.add(event);if (!_hasPending) {_state |= _STATE_HAS_PENDING;if (!_isPaused) {pending.schedule(this);}}}

这个方法主要是添加一个处理事件，接着我们看看schedule方法

   void schedule(_EventDispatch<T> dispatch) {if (isScheduled) return;assert(!isEmpty);if (_eventScheduled) {assert(_state == stateCanceled);_state = stateScheduled;return;}scheduleMicrotask(() {int oldState = _state;_state = stateUnscheduled;if (oldState == stateCanceled) return;handleNext(dispatch);});_state = stateScheduled;}放到微任务中执行handleNextvoid handleNext(_EventDispatch<T> dispatch) {assert(!isScheduled);assert(!isEmpty);_DelayedEvent event = firstPendingEvent!;_DelayedEvent? nextEvent = event.next;firstPendingEvent = nextEvent;if (nextEvent == null) {lastPendingEvent = null;}event.perform(dispatch);}

_DelayedEvent是个抽象类，perform方法是需要子类去实现
它的子类就是_DelayedData

class _DelayedData<T> extends _DelayedEvent<T> {final T value;_DelayedData(this.value);void perform(_EventDispatch<T> dispatch) {dispatch._sendData(value);}
}

我们看_sendData这个方法，这个方法也是在_BufferingStreamSubscription类中实现的。
整个代码看下来我们只需要关心 _zone.runUnaryGuarded(_onData, data); 这个是实现数据发送的关键。
也是订阅者和被订阅者之间的中间角色，起着承上启下的作用。

 void _sendData(T data) {assert(!_isCanceled);assert(!_isPaused);assert(!_inCallback);bool wasInputPaused = _isInputPaused;_state |= _STATE_IN_CALLBACK;_zone.runUnaryGuarded(_onData, data);_state &= ~_STATE_IN_CALLBACK;_checkState(wasInputPaused);}

Zone 被称做是沙箱，在 Dart 中，Zone 是用于隔离代码执行环境的概念。Zone 可以看作是一种执行上下文，它可以用于控制代码执行过程中的一些行为，比如异常处理、日志记录、资源管理等。Zone 中的 run 方法允许你在指定的 Zone 中运行一段代码块。
这里不继续深究下去，有兴趣的可以看看。我们只需要知道调用runUnaryGuarded这个方法。它就会回调_onData函数，并将参数data传给_onData方法。
_onData这个函数就是我们listen监听里的匿名函数，最终会回调那里去。