深耕Flutter---扫清Flutter异步编程之路上的路障

Isolate

基本概念

Isolate更像是一个进程,全部的dart代码都会在上面运行,其内部有一个线程,用来处理event loop,还有一块内存,而且这块内存是私有的,也就说两个Isolate不共享内存(这也是和Java 的Thread的主要区别),要是多个Isolate想共同工做,只能经过port来发送消息进行通讯(通常会把特别耗时的操做,好比IO操做、图片压缩等容易引发main isolate卡顿的操做放到一个单独的Isolate中运行)。这样作的好处是分配内存和垃圾回收的时候不用对操做加锁,对于像flutter这种须要快速构建或者拆卸大量widget的应用来讲颇有帮助。
两个独立运行的Isolatejava

建立Isolate

两种方式建立:web

  1. 经过静态方法Isolate.spawn建立(经常使用)
external static Future<Isolate> spawn<T>(
      void entryPoint(T message), T message,
      { 
 
  bool paused: false,
      bool errorsAreFatal,
      SendPort onExit,
      SendPort onError,
      @Since("2.3") String debugName});

entryPoint入参必须是一个顶层函数或者一个static方法,不接受一个函数表达式或者一个实例的方法,不然会Invalid argument(s): Isolate.spawn expects to be passed a static or top-level function的错误。算法

  1. 经过构造函数建立(几乎用不到)
Isolate(this.controlPort, { 
 
  this.pauseCapability, this.terminateCapability});

使用这种方法建立Isolate必须本身传入pause和terminate能力,不然调用pausekill方法是没有任何效果的。编程

Isolate间通讯

  • 不一样的Isolate之间通讯是经过 ReceivePortSendPort来实现的。
  • 能够在Isoalte之间传输的数据类型有基础类型(null, num, bool, double,String)、SendPortListMapListMap元素的类型也必须是容许传输的类型,容许传递SendPort是实现双向通讯的基础),实践中发现也能传递一些属性为基础类型的简单对象,在1.17.1版本中甚至能传递future对象。
  • ReceivePort 实现了Stream类,是一个非广播的Stream,因此只容许有一个监听者。
  • 经过看C的源码能够知道,每一个Isolate其实对应一个线程,其背后是有一个线程池的。
  • 建立Isolate时会建立对应的堆空间、消息处理handler等等对应的信息。

单向通讯

void startOnWay() async { 
 
  
  ReceivePort receivePort = ReceivePort();
  Isolate isolate = await Isolate.spawn(oneWayEntryPoint, receivePort.sendPort);
  receivePort.listen((data) { 
 
  
    print('$data');
    receivePort.close();//若是不关的,它会一直监听
    isolate?.kill(priority: Isolate.immediate);//杀死isolate
    isolate = null;
  });
}

void oneWayEntryPoint(SendPort sendPort) { 
 
  
  sendPort.send('hello world!');
}

双向通讯

void startBothWay() async { 
 
  
  ReceivePort receivePort = ReceivePort();

  await Isolate.spawn(entryPointBothWay, receivePort.sendPort);

  receivePort.listen((data) { 
 
  
    if (data is SendPort) { 
 
  
      data.send('hello child!');
    } else { 
 
  
      print('parent: $data');
    }
  });
}

void entryPointBothWay(SendPort sendPort) { 
 
  
  ReceivePort r = ReceivePort();
  sendPort.send(r.sendPort);

  r.listen((data) { 
 
  
    print('child: $data');
    sendPort.send('hello parent!');
  });
}

compute函数

在flutter中实现isolate间通讯,能够直接使用更方便的API–compute函数。它其实是对isolate之间通讯的封装,每次调用后都会执行isolate.kill方法。demo以下:api

class Food { 
 
  
  String name;
  double price;

  Food(this.name, this.price);
}

class AsyncDemo extends StatefulWidget { 
 
  
  @override
  _AsyncState createState() => _AsyncState();
}

class _AsyncState extends State<AsyncDemo> { 
 
  
  var food = Food('apple', 23.1);

  @override
  Widget build(BuildContext context) { 
 
  
    return Scaffold(
      body: Center(
        child: Text(
          '''
          name:   ${ 
 
  food.name}
          price:  ${ 
 
  food.price}
          ''',
          style: TextStyle(
            fontSize: 22.0,
          ),
        ),
      ),
      floatingActionButton: FloatingActionButton(onPressed: () { 
 
  
        compute<Food, Food>(handleCounter, food).then((value) { 
 
  
          setState(() { 
 
  
            food = value;
          });
        });
      }),
    );
  }
}

Food handleCounter(Food food) { 
 
  
  return food..price = 1.8;
}

EventLoop

经过Isolate咱们知道dart app是个单线程app,那他是怎么实现异步操做的呢?答案是EventLoop和他相关联的两个队列–event queue、microtask queue。缓存

Event Queue

event queue包含几乎全部的事件,好比IO操做、绘制操做、多个isolate之间通讯、Future的部分操做等等。数据结构

Microtask Queue

这个队列用来执行用时很短的事件,这些事件会在把控制权交给Event Queue以前运行完毕。在实际开发中使用场景十分有限,整个Flutter源码里面总共有7处使用到了Microtask Queue(flutter中将事件提交到Microtask Queue上是使用 scheduleMicrotask api来操做的)。app

执行顺序

执行顺序
从上图能够看到,老是会先执行microtask,当microtask queue空的时候才会去执行event queue里的事件,每执行完一个event,都会从新去执行microtask queue里的事件。若是执行microtask里的事件消耗太多的时间,容易形成event queue阻塞,从而形成app的卡顿。
看下面demo,less

import 'dart:async';
main() { 
 
  
  print('main #1 of 2');
  scheduleMicrotask(() => print('microtask #1 of 3'));

  new Future.delayed(new Duration(seconds:1),// 1
      () => print('future #1 (delayed)'));

  new Future(() => print('future #2 of 4')) // 2
      .then((_) => print('future #2a'))
      .then((_) { 
 
  
        print('future #2b');
        scheduleMicrotask(() => print('microtask #0 (from future #2b)'));
      })
      .then((_) => print('future #2c')); // X

  scheduleMicrotask(() => print('microtask #2 of 3'));

  new Future(() => print('future #3 of 4')) // 3
      .then((_) => new Future(
                   () => print('future #3a (a new future)')))// 5
      .then((_) => print('future #3b'));// Y

  new Future(() => print('future #4 of 4')); // 4
  scheduleMicrotask(() => print('microtask #3 of 3'));
  print('main #2 of 2');
}

输出:异步

main #1 of 2
main #2 of 2 // a
microtask #1 of 3
microtask #2 of 3
microtask #3 of 3 // b
future #2 of 4
future #2a
future #2b // c
future #2c
microtask #0 (from future #2b)
future #3 of 4
future #4 of 4
future #3a (a new future)
future #3b
future #1 (delayed)

分析
程序会从上到下一次执行main方法里的同步方法,因此会依此输出到标 a的位置。scheduleMicrotask会把任务提交到microtask queue里,future会被提交到event queue里,因此先执行scheduleMicrotask的操做,对应输出中标记 b 的地方。microtaks queue里的事件都被处理完了,接下来,eventloop开始处理event queue里的事件。按道理来说代码里标记 1 2 3 4的future会被依此放到event queue队列中,可是标 1 的地方有一秒的延迟,因此标 1 的地方会在一秒后才会被放到event queue里,因此此时event queue里的事件排序是:2 3 4。先执行 2,依此输出到标记 c 的地方。这个时候看到了一个scheduleMicrotask,因此会把这个事件放到microtask的队列中去,此时这个future还未执行完,继续执行,输出future #2c。执行完毕后,event queue里的事件变成了 3 4,回去接着执行microtask queue里的事件,输出microtask #0 (from future #2b)。此时microtask queue里没有事件了,接着执行event queue里的事件,也就是事件 3,输出future #3 of 4,当执行到 5 的时候,又生成了一个新的future,放到了event queue里,事件3也就执行完了,此时event queue里的事件变成了 4(等待执行) 5(等待执行)。接着执行事件4,输出future #4 of 4,接着执行事件 5,依此输出future #3a (a new future) future #3b,最后通过一秒后事件1入队被执行,也就是最后输出的future #1 (delayed)。
须要注意的是代码中标记X和Y的地方,标记X的地方,因为他的上一步并无返回Future,因此标X的操做是在事件2后面进行的,而标记Y的地方,他上一步返回了一个Future对象,也就是事件 5,因此这个then函数里的操做是跟在事件5后面进行的。

Future

按官方文档上的介绍:它表明在未来某一时刻会获取到想要的值。当一个返回Future的函数被调用的时候,会发生两件事情:

  1. 这个Future对象会进去队列中排队等待被执行,此时会直接返回一个未完成状态的Future对象,可是这个函数是不进去队列的,它仅仅是被调用而已。
  2. 当值可得到的时候,Future处于一个叫 Completed with value 的状态,当执行失败时,处于一个叫作Completed with error的状态。

Future的三种状态:

在这里插入图片描述

要想使用Future,能够直接使用其提供的api,也可使用dart提供的语法糖async、await。

async、await

main() { 
 
  
  fetchData();
}

void fetchData() async { 
 
  
  print('fetch data start');
  await Future.delayed(Duration(seconds: 2));
  print('Data is returned!');
}

//输出:
//fetch data start
//Data is returned!(两秒后会输出)

须要注意的是,在dart 2.0 以前,async中的同步代码不会运行,碰见async就直接返回一个未完成的Future对象。当被async标记的函数没有返回值时,会返回一个Future< void>对象,有返回值的话,返回一个Future< T>对象。

Future API

Future()

void main() { 
 
  
  var future = Future((){ 
 
  
    return 'hello world!';
  });

  future.then(print);
}

背后是经过Timer来实现的,直接看源码:

factory Future(FutureOr<T> computation()) { 
 
  
  _Future<T> result = new _Future<T>();
  Timer.run(() { 
 
  
    try { 
 
  
      result._complete(computation());
    } catch (e, s) { 
 
  
      _completeWithErrorCallback(result, e, s);
    }
  });
  return result;
}

Future.delayed

Future.delayed(Duration(seconds: 1), () { 
 
  
    print('delayed 1');
  });

1秒后输出相应值,其背后也是用的timer来实现,只不是传了一个duration对象过去。

factory Future.delayed(Duration duration, [FutureOr<T> computation()]) { 
 
  
  _Future<T> result = new _Future<T>();
  new Timer(duration, () { 
 
  
    if (computation == null) { 
 
  
      result._complete(null);
    } else { 
 
  
      try { 
 
  
        result._complete(computation());
      } catch (e, s) { 
 
  
        _completeWithErrorCallback(result, e, s);
      }
    }
  });
  return result;
}

Future.value

Future.value(1).then(print);

背后调用的源码以下(只考虑value非Future的状况下):

void _asyncCompleteWithValue(T value) { 
 
  
  _setPendingComplete();
  _zone.scheduleMicrotask(() { 
 
  
    _completeWithValue(value);
  });
}

能够看到背后并无调用Timer相关api,使用的是scheduleMicrotask,也就是说Future.value函数是在mictotask queue里完成的,能够经过下面的例子验证:

Future.delayed(Duration(seconds: 0)).then((value) => print('Future.delayed'));
Future.value('Future.value').then(print);

//输出:
//Future.value ---优先执行microtask queue里的事件。
//Future.delayed

Future.sync

Future.sync(() => 'sync').then(print);

同步执行代码,下面是源码实现:

factory Future.sync(FutureOr<T> computation()) { 
 
  
  try { 
 
  
    var result = computation();
    if (result is Future<T>) { 
 
  
      return result;
    } else { 
 
  
      return new _Future<T>.value(result as dynamic);
    }
  } catch (error, stackTrace) { 
 
  
    var future = new _Future<T>();
    AsyncError? replacement = Zone.current.errorCallback(error, stackTrace);
    if (replacement != null) { 
 
  
      future._asyncCompleteError(replacement.error, replacement.stackTrace);
    } else { 
 
  
      future._asyncCompleteError(error, stackTrace);
    }
    return future;
  }
}

要注意的是若是computation返回一个非future对象,会经过Future.value建立对象Future对象并返回,别忘了Future.value是在microtask 里完成的。再对比下下面两个例子的输出:

Demo1:
Future.delayed(Duration(seconds: 0)).then((value) => print('delayed'));
Future.sync(() => Future.delayed(Duration(seconds: 0)))
    .then((value) => print('sync'));
//输出:
//delayed
//sync

Demo2:
Future.delayed(Duration(seconds: 0)).then((value) => print('delayed'));
Future.sync(() => 'sync').then(print);
//输出:
//sync
//delayed

Future.wait

final f1 = Future.delayed(Duration(seconds: 1)).then(
  (value) => 'first future.',
);

final f2 = Future.delayed(Duration(seconds: 2)).then(
  (value) => 'second future.',
);

Future.wait([f1, f2]).then((value) => print(value));
//输出:
//[first future., second future.]

等待全部的future完成并返回一个包含全部future执行结果的list,一旦有一个future执行过程当中出现了error,最后的结果都不会输出。源码实现:

static Future<List<T>> wait<T>(Iterable<Future<T>> futures,
    { 
 
  bool eagerError: false, void cleanUp(T successValue)}) { 
 
  
  final _Future<List<T>> result = new _Future<List<T>>();
  List<T> values; // Collects the values. Set to null on error.
  int remaining = 0; // How many futures are we waiting for.
  var error; // The first error from a future.
  StackTrace stackTrace; // The stackTrace that came with the error.

  // Handle an error from any of the futures.
  // TODO(jmesserly): use `void` return type once it can be inferred for the
  // `then` call below.
  handleError(Object theError, StackTrace theStackTrace) { 
 
  
    remaining--;
    if (values != null) { 
 
  
      if (cleanUp != null) { 
 
  
        for (var value in values) { 
 
  
          if (value != null) { 
 
  
            // Ensure errors from cleanUp are uncaught.
            new Future.sync(() { 
 
  
              cleanUp(value);
            });
          }
        }
      }
      values = null;
      if (remaining == 0 || eagerError) { 
 
  
        result._completeError(theError, theStackTrace);
      } else { 
 
  
        error = theError;
        stackTrace = theStackTrace;
      }
    } else if (remaining == 0 && !eagerError) { 
 
  
      result._completeError(error, stackTrace);
    }
  }

  try { 
 
  
    // As each future completes, put its value into the corresponding
    // position in the list of values.
    for (var future in futures) { 
 
  
      int pos = remaining;
      future.then((T value) { 
 
  
        remaining--;
        if (values != null) { 
 
  
          values[pos] = value;
          if (remaining == 0) { 
 
  
            result._completeWithValue(values);
          }
        } else { 
 
  
          if (cleanUp != null && value != null) { 
 
  
            // Ensure errors from cleanUp are uncaught.
            new Future.sync(() { 
 
  
              cleanUp(value);
            });
          }
          if (remaining == 0 && !eagerError) { 
 
  
            result._completeError(error, stackTrace);
          }
        }
      }, onError: handleError);
      // Increment the 'remaining' after the call to 'then'.
      // If that call throws, we don't expect any future callback from
      // the future, and we also don't increment remaining.
      remaining++;
    }
    if (remaining == 0) { 
 
  
      return new Future.value(const []);
    }
    values = new List<T>(remaining);
  } catch (e, st) { 
 
  
    // The error must have been thrown while iterating over the futures
    // list, or while installing a callback handler on the future.
    if (remaining == 0 || eagerError) { 
 
  
      // Throw a new Future.error.
      // Don't just call `result._completeError` since that would propagate
      // the error too eagerly, not giving the callers time to install
      // error handlers.
      // Also, don't use `_asyncCompleteError` since that one doesn't give
      // zones the chance to intercept the error.
      return new Future.error(e, st);
    } else { 
 
  
      // Don't allocate a list for values, thus indicating that there was an
      // error.
      // Set error to the caught exception.
      error = e;
      stackTrace = st;
    }
  }
  return result;
}

总体来讲很简单,能够看到内部维护了一个future的列表,当所有执行完毕后返回结果,若是出错就直接返回错误信息。

Future.microtask

开启一个microtask的简单写法,用法同scheduleMicrotask,内部也是经过scheduleMicrotask来实现的,直接看源码吧:

factory Future.microtask(FutureOr<T> computation()) { 
 
  
  _Future<T> result = new _Future<T>();
  scheduleMicrotask(() { 
 
  
    try { 
 
  
      result._complete(computation());
    } catch (e, s) { 
 
  
      _completeWithErrorCallback(result, e, s);
    }
  });
  return result;
}

异常处理

async、await的异常处理

async、await的异常处理和同步代码的异常处理是同样的,采用try catch的方法,一样也是可使用finally的:

main() { 
 
  
  errorDemo();
}

void errorDemo() async { 
 
  
  try { 
 
  
    await Future.delayed(Duration(seconds: 1)).then(
      (value) => throw '####Error####',
    );
  } catch (e) { 
 
  
    print(e);
  } finally { 
 
  
    print('finally compeleted.');
  }
}
// 输出:
// ####Error####
// finally compeleted.

Future API 的异常处理

方法简介

Future api提供两种方式来处理异常,一个是Future<R> then<R>(FutureOr<R> onValue(T value), {Function onError}); 里的onError回调,另外一个是Future<T> catchError(Function onError, {bool test(Object error)});

  • 使用onError回调

    main() { 
     
        
      Future.delayed(Duration(seconds: 1), () { 
     
        
        throw '#####Error#####';
      }).then(print, onError: (error) { 
     
        
        print('has error');
      }).whenComplete(() => print('has complete!'));
    }
    // 输出:
    // has error
    // has complete!

    能够看到并无报错,而是按照预期依此调用了onError、whenCompelte(至关于try catch里的 finally)回调。这有一个小细节须要注意的是,onError和onValue是同级的,当onValue里面发生异常时是不会走同级的onError回调的,onValue和onError是服务于一个future的。

  • 使用catchError

    Future.delayed(Duration(seconds: 1), () { 
     
        
      throw '#####Error#####';
    }).catchError((error){ 
     
        
      print('has error');
    });
    // 输出:
    // has error

    catchError和onError回调的用法差很少,最后也是输出了has error。看他的api,发现还有一个叫作test的可选命名参数,它是作什么的呢?只有当test返回true的时候才会走catchError回调,返回为false时会直接报错。他的入参是error对象,因此能够在test里面作一些特殊处理,好比error类型的过滤等等,默认状况下test始终返回true。

    main() { 
     
        
      Future.delayed(Duration(seconds: 1), () { 
     
        
        throw '#####Error#####'; // 1
      }).catchError((String error) { 
     
        
        print('has error');
      }, test: (error) { 
     
        
        if (error is String) { 
     
        
          return true;
        }
        return false;
      });
    }

    能够把标1的地方改为 throw 1再运行下试试效果。

onError和catchError的优先级问题

main() { 
 
  
  Future.delayed(Duration(seconds: 1), () { 
 
  // 1
    throw '##### Error #####'; // 2
  }).then(print, onError: (error) { 
 
   // 3
    print('onError callback'); // 4
  }).catchError((error) { 
 
   
    print('catchError callback');
  });
}
// 输出:
// onError callback.

运行完delayed方法后,会返回一个Future,叫他Future1,标记 3 处的onError里的回调只对Future 1负责,Future 1爆了异常因此会走到onError回调里,也就是标记 4 的地方,一旦onError捕获了异常就不会再去执行catchError里的代码。若是说标 3 的地方并无设置onError回调,才会往catchError回调里走。

链式调用的异常处理

链式调用中,一旦某个环节爆了异常,下面的每个节点都会返回相同的错误,直到遇到catchError回调或者被下面的某个节点的onError回调拦截,直接看官网的一个例子吧:

Future<String> one() => new Future.value("from one");

Future<String> two() => new Future.error("error from two");

Future<String> three() => new Future.value("from three");

Future<String> four() => new Future.value("from four");

void main() { 
 
  
  one() // Future completes with "from one".
      .then((_) => two()) // Future completes with two()'s error.
      .then((_) => three()) // Future completes with two()'s error.
      .then((_) => four()) // Future completes with two()'s error.
      .then((value) => Future.value(1)) // Future completes with two()'s error.
      .catchError((e) { 
 
  
    print("Got error: ${e.error}"); // Finally, callback fires.
    return 42; // Future completes with 42.
  }).then((value) { 
 
  
    print("The value is $value");
  });
}

// Output of this program:
// Got error: error from two
// The value is 42

同步、异步混合异常处理

不管是async、await,仍是Future,都是异步执行的,上面介绍的错误处理方式只能处理异步的异常,那么同步代码和异步代码混合的状况下该怎么处理异常呢?看下面的例子:

void main() { 
 
  
  getLegalErpLength().then(print).catchError((error) { 
 
  
    print('error happen!');
  });
}

Future<int> getLegalErpLength() { 
 
  
  final erp = findErpFromDb(); // 1
  return Future.value(erp).then(parseErp);
}

int parseErp(String erp) { 
 
  
  return erp.length;
}

String findErpFromDb() { 
 
  
  throw 'unknown error';
}

//输出:
Unhandled exception:
unknown error
#0      findErpFromDb 
#1      getLegalErpLength 
#2      main
#3      _startIsolate.<anonymous closure> (dart:isolate-patch/isolate_patch.dart:301:19)
#4      _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:168:12)

Process finished with exit code 255

getLegalErpLength是一个异步方法,里面调用了同步方法:findErpFromDb,当findErpFromDb出现异常的时候getLegalErpLength时没法进行捕获的,因此直接输出了上面的错误栈。那若是想捕获findErpFromDb的异常应该怎么写呢?须要引入Future.sync ,修改以下:

Future<int> getLegalErpLength() { 
 
  
  return Future.sync(() { 
 
  
    final erp = findErpFromDb();
    return Future.value(erp).then(parseErp);
  });
}
// 输出:
// error happen!

使用Future.sync将同步方法和异步方法包装一下,这样就完美的解决了上面的问题。

Future原理简介

  • Future()、Future.delayed()

    背后使用的timer来实现,具体的源码能够参考dart sdk中的timer_impl.dart文件。简单来说,是用一个_TimerHeap的数据结构来存取timer,内部采用堆排序算法对其进行排序,当预先设置的时间到时,有一个eventhandler的东西负责处理设置的回调。

  • Future.value()、Future.microtask()

    内部层层调用后,最终调用scheduleMicrotask方法,将其放到microtask queue中执行。内部也维护了一个loop,将每个回调封装成_AsyncCallbackEntry放到loop中,而后循环执行,直到loop为空。具体能够参考schedule_microtask.dart文件。

建立Future的另一种方式:Completer

completer能够用来建立future,相比使用future,completer能够本身指定future的完成时机。在开发中,能使用future的地方就直接使用future,它比completer更易读,除非有些状况下拿不到想要的future对象,可使用completer来替代,好比下面这个获取图片长宽的例子:

class CompleterDemo extends StatelessWidget { 
 
  
  Widget build(BuildContext context) { 
 
  
    Image image = new Image.network('https://img01.e23.cn/2020/0519/20200519022445869.jpg');
    Completer<ui.Image> completer = new Completer<ui.Image>();
    image.image.resolve(new ImageConfiguration()).addListener(ImageStreamListener(
      (image, synchronousCall) { 
 
  
        completer.complete(image.image);
      },
    ));
    return new Scaffold(
      appBar: new AppBar(
        title: new Text("Image Dimensions Example"),
      ),
      body: new ListView(
        children: [
          new FutureBuilder<ui.Image>(
            future: completer.future,
            builder: (BuildContext context, AsyncSnapshot<ui.Image> snapshot) { 
 
  
              if (snapshot.hasData) { 
 
  
                return new Text(
                  '${snapshot.data.width}x${snapshot.data.height}',
                  style: Theme.of(context).textTheme.display3,
                );
              } else { 
 
  
                return new Text('Loading...');
              }
            },
          ),
          image,
        ],
      ),
    );
  }
}

Stream

相比future,stream能够用来处理数据流。Stream分为两种,一种是single subscription streams,只容许监听一次,另一种是Broadcast streams,能够容许监听屡次。直接看Stream的API你会发现,这东西和RxJava实在是太像了,一些API的名字和用法几乎差很少,因此这里再也不对Stream的经常使用API进行梳理,感兴趣的直接去看官方文档吧,对RxJava有了解的同窗学起来是很容易的。

建立Stream

  • Stream自带的一些factory构造函数,好比Stream.fromFuture(Future future)、Stream.fromIterable(Iterable elements)等等。以fromIterable为例:

    Stream.fromIterable([1, 2]).listen((event) { 
     
        
      print('event is $event');
    });
    
    // 输出:
    // event is 1
    // event is 2
  • 异步生成器函数:async* and yield(那对应的同步生成器函数使用的是 sync* and yield)

    main() async { 
     
        
      print('Start...');
      await for (int number in generateNumbers(2)) { 
     
         // 1
        print(number);
      }
      print('End');
    }
    
    Stream<int> generateNumbers(int maxCount) async* { 
     
        
      print('yield output: ');
      for (int i = 0; i < maxCount; i++) { 
     
        
        yield i; // 2
      }
    
      print('yield* output: ');
      yield* createNumbers(maxCount); // 3
    }
    
    Stream<int> createNumbers(int maxCount) async* { 
     
        
      for (int i = 0; i <= maxCount; i++) { 
     
        
        yield i;
      }
    }
    
    
    // 输出:
    // Start...
    // yield output: 
    // 0
    // 1
    // yield* output: 
    // 0
    // 1
    // 2
    // End

    须要注意两点:

    一、异步生成器的产物Stream可使用await for进行遍历,await for必须实在async标记的函数中才会有效,另外同步生成器的产物Iterable可使用for进行遍历。

    二、代码中标记2的地方使用的是 yield,标记3的地方使用的是yield*,他俩的区别是一个生成单个值,一个生成一串值,也就是一个Stream。

  • 使用StreamController建立Stream

    使用StreamController建立Stream能够在任何地方、任什么时候间添加event并处理event,可是和async相比要复杂的多,最重要的一点是async建立的Stream不会立刻执行,当第一个注册者注册的时候才会执行,可是利用StreamController建立的Stream则否则,看下面的代码:

    main() async { 
     
        
      var counterStream = timedCounter(const Duration(seconds: 1), 5);
      await Future.delayed(Duration(seconds: 3), () { 
     
        
        print('3 seconds has gone.');
      });
    	counterStream.listen((value) { 
     
        
        print('$value -- ${DateTime.now()}');
      }); 
    }
    
    Stream<int> timedCounter(Duration interval, [int maxCount]) { 
     
        
      var controller = StreamController<int>();
      int counter = 0;
      void tick(Timer timer) { 
     
        
        counter++;
        controller.add(counter); // Ask stream to send counter values as event.
        if (maxCount != null && counter >= maxCount) { 
     
        
          timer.cancel();
          controller.close(); // Ask stream to shut down and tell listeners.
        }
      }
    
      Timer.periodic(interval, tick); // BAD: Starts before it has subscribers.
      return controller.stream;
    }
    // 输出:
    3 seconds has gone.
    1 -- 2020-05-19 21:52:20.843943
    2 -- 2020-05-19 21:52:20.847416
    3 -- 2020-05-19 21:52:20.847438
    4 -- 2020-05-19 21:52:21.831694
    5 -- 2020-05-19 21:52:22.828021

能够看到前三个几乎是同时输出的,这是由于StreamController一旦建立Stream,里面的代码就开始执行,生成的数据会缓存下来,当有注册的时候立马把缓存输出给监听者,为了不这种状况,可使用StreamController各类回调方法进行限制,上面的代码修改以下:

Stream<int> timedCounter(Duration interval, [int maxCount]) { 
 
  
  StreamController controller;
  int counter = 0;
  Timer timer;

  void tick(_) { 
 
  
    counter++;
    controller.add(counter);
    if (maxCount != null && counter >= maxCount) { 
 
  
      timer.cancel();
      controller.close();
    }
  }

  void startTimer() { 
 
  
    timer = Timer.periodic(interval, tick);
  }

  controller = StreamController<int>(
    onListen: () { 
 
  
      startTimer();
    },
  );
  return controller.stream;
}

//输出:
3 seconds has gone.
1 -- 2020-05-19 22:02:32.488114
2 -- 2020-05-19 22:02:33.479319
3 -- 2020-05-19 22:02:34.477549
4 -- 2020-05-19 22:02:35.476372
5 -- 2020-05-19 22:02:36.481389

StreamController还有onCancel、onPause、onResume三个回调,用来处理取消、暂停、继续等操做。

Stream原理简介

Stream背后实现采用的是scheduleMicrotask,也就是说经过Stream发送的事件会直接被添加到microtask event queue里面。下面来跟踪下源码:

//StreamController里定义的add方法:
void add(T event);

//具体在_StreamController里面实现:
void _add(T value) { 
 
  
    if (hasListener) { 
 
  
      _sendData(value);
    } else if (_isInitialState) { 
 
  
      _ensurePendingEvents().add(new _DelayedData<T>(value));
    }
}

//容许建立同步的和异步的StreamController,咱们直接来看异步的实现:
abstract class _AsyncStreamControllerDispatch<T>
    implements _StreamController<T> { 
 
  
  void _sendData(T data) { 
 
  
    _subscription._addPending(new _DelayedData<T>(data));
  }
}

//继续往里跟,会首先放到pending里面,而后判断是否为暂停状态,若是不是的话直接调用pending的schedule方法
void _addPending(_DelayedEvent event) { 
 
  
    _StreamImplEvents<T> pending = _pending;
    if (_pending == null) { 
 
  
      pending = _pending = new _StreamImplEvents<T>();
    }
    pending.add(event);
    if (!_hasPending) { 
 
  
      _state |= _STATE_HAS_PENDING;
      if (!_isPaused) { 
 
  
        _pending.schedule(this);
      }
   }
}
//schedule的实现,看到了熟悉的scheduleMicrotask方法。
  void schedule(_EventDispatch<T> dispatch) { 
 
  
    if (isScheduled) return;
    assert(!isEmpty);
    if (_eventScheduled) { 
 
  
      assert(_state == _STATE_CANCELED);
      _state = _STATE_SCHEDULED;
      return;
    }
    scheduleMicrotask(() { 
 
  
    //先判断状态,若是已经取消了,则中止处理。
      int oldState = _state;
      _state = _STATE_UNSCHEDULED;
      if (oldState == _STATE_CANCELED) return;
      handleNext(dispatch);
    });
    _state = _STATE_SCHEDULED;
  }

总结

异步编程在平时开发中十分常见,了解原理的重要性没必要多说,因此直接点个赞吧,双击六六六!

参考资料: https://web.archive.org/web/20170704074724/https://webdev.dartlang.org/articles/performance/event-loop https://web.archive.org/web/20170702215338/https://www.dartlang.org/articles/libraries/futures-and-error-handling https://web.archive.org/web/20170702213650/https://www.dartlang.org/tutorials/language/futures https://dart.dev/codelabs/async-await https://medium.com/dartlang/dart-asynchronous-programming-futures-96937f831137 https://medium.com/dartlang/dart-asynchronous-programming-isolates-and-event-loops-bffc3e296a6a

相关文章
相关标签/搜索