Dart Asynchronous support

Dart is a single-threaded model, so there is no main thread/child thread separation. Dart is also an Event-Looper and event-Queue, where all events are executed in sequence by the **EventLooper EventLooper

The Dart’s Event Loop

• Get the Event from the EventQueue
• Handling the Event
• Until the EventQueue is empty

Event Queue

Single threaded model

Once a Dart function is executed, it will execute until the end of the function, that is, the Dart function will not be interrupted by other Dart code

Dart doesn’t have threads, it just has isolates, and each isolate doesn’t share memory and a Dart program starts with the Main function of the Main isolate, and after the Main function ends, Main ISOLATE Threads start processing each Event in the Event Queue one by one Main Isolate

Why Microtask Queue exists

This Queue is intended to process things that are later, but need to be processed before the next message arrives

Asynchronous task scheduling

Dart: Async: dart:async: dart:async: Dart :async: Dart :async: Dart :async

• Using the Future class, you can add tasks to the end of the Event Queue
• Add tasks to the end of the Microtask Queue using the scheduleMicrotask function

When using EventQueue, you need to be careful not to make the MicroTask Queue too large, or it will block the processing of other events Use Event Queue

Future

A Future is a class that implements asynchracy, similar to the Promise in JS

Future<T> Future(FutureOr<T> Function() computation)
Copy the code

main(List<String> args) {
  var futureInstance = Future<String> (() = >"12345");
  futureInstance.then((res) {
    print(res);
  }).catchError((err) {
    print(err);
  });
}
Copy the code

Future.value

main(List<String> args) {
var futureInstance = Future.value(1234);
futureInstance.then((val)=>print(val));
}
// Result 1234 is displayed
Copy the code

main(List<String> args) {
  var futureInstance = Future.error("This is a mistake.");
  futureInstance.catchError((err) {
    print(err);
  });
}

This is an error
Copy the code

Future.then

The future. then function is commonly used when a divide-and-conquer task needs to be split into many smaller tasks to be executed step by step

void main() {
  new Future(() => futureTask) // Function of asynchronous task
      .then((m) => {print("futueTask execute result:$m")}) // Subtasks after the task is executed
      .then((m){print(m.length);return 32; })// where m is the result returned after the execution of the previous task
      .then((m) => printLength(m))
      .whenComplete(() => whenTaskCompelete); // A callback function when all tasks are completed

  futureTask();
}

int futureTask() {
  return 21;
}

void printLength(int length) {
  print("Text Length:$length");
}

void whenTaskCompelete() {
  print("Task Complete");
}
Copy the code

[Running] dart "/Users/kim/test/main.dart"
futueTask execute result:Closure: () => int from Function 'futureTask': static.
1
Text Length:32
[Done] exited with code=0 in 0.209 seconds
Copy the code

Future.delayed(const Duration(seconds: 1), () => futureTask);
Copy the code

import 'dart:async';

main(List<String> args) {
  //———————————————— method 1 ————————————————
  Future delay(int ms) {
    var com = Completer();
    Timer(Duration(milliseconds: ms), () {
      com.complete("1");
    });
    return com.future;
  }

  delay(1000).then((res) {
    print(res);
  });

  //———————————————— method 2 ————————————————
  Future.delayed(Duration(milliseconds: 2000), () {
    print("Method 2");
  });
}
Copy the code

Future. The add and the Future. Wait

main(List<String> args) {
  Future future1() async { print("This is asynchronous function one."); }
  Future future2() async { print("This is asynchronous function two."); }

  var futureList = <Future>[];
  futureList.add(future1());
  futureList.add(future2());
  Future.wait(futureList);
}
Copy the code

Future.sync((){

});
Copy the code

Future.microtask

The scheduleMicrotask function is used to add the current task to the microtask-queue

main(List<String> args) {
  Future.microtask(() {
    // Prioritized business logic
  });
}
Copy the code

async.scheduleMicrotask(() => microtask());

void microtask(){
  // doing something
}
Copy the code

To deal with the Future

You can obtain the result of a Future execution in two ways:

• Use async and await
• Use the Future API for detailed description, refer to the library overview

Code that uses the async and await keywords is asynchronous. Although it seems a little bit like trying to synchronize your code. For example, the following code uses await to wait for the execution result of an asynchronous function.

await lookUpVersion();
Copy the code

To use await, the code must be in an asynchronous function (a function that uses the async flag) :

Future checkVersion() async {
  var version = await lookUpVersion();
  // Do something with version
}
Copy the code

Tip: While an asynchronous function may perform time-consuming operations, it does not wait for them. In contrast, asynchronous functions are executed only when they encounter the first await expression (see more on this). That is, it returns a Future object and resumes execution only after the await expression is completed.

Use try, catch, and finally to handle errors caused by using await in code.

try {
  version = await lookUpVersion();
} catch (e) {
  // React to inability to look up the version
}
Copy the code

You can use await multiple times in an asynchronous function. For example, the following code waits for the result of a function three times:

var entrypoint = await findEntrypoint();
var exitCode = await runExecutable(entrypoint, args);
await flushThenExit(exitCode);
Copy the code

In await expressions, the value of the expression is usually a Future object; If not, the value of the expression is automatically wrapped as a Future object

• The Future object specifies a promise to return an object (similar to a promise)
• The result of execution of the await expression is the returned object.
• Await expressions block the execution of code until the desired object is returned.

If using await results in a compile-time error, verify that await is in an asynchronous function. For example, to use await in the application’s main() function, the body of the main() function must be marked async:

Future main() async {
  checkVersion();
  print('In main: version is ${await lookUpVersion()}');
}
Copy the code

Declaring asynchronous functions

A function whose body is marked by an async identifier is an asynchronous function. Add the async keyword to the function to return the Future. For example, consider the following synchronization function, which returns a String:

String lookUpVersion() => '1.0.0';
Copy the code

For example, if a Future implementation would be time-consuming, change it to an asynchronous function that returns a Future.

Future<String> lookUpVersion() async= >'1.0.0';
Copy the code

Note that the function body does not need to use the Future API. Dart creates Future objects if necessary. If the function does not return a valid value, set its return type to Future<void>.

await for (varOrType identifier in expression) {
  // Executes each time the stream emits a value.
  // executes each time the stream emits a value
}
Copy the code

Future main() async {
  // ...
  await for (var request in requestServer) {
    handleRequest(request);
  }
  // ...
}
Copy the code

Stream

A Stream is very idiosyncratic but not very easy to understand, and I prefer to think of it as a factory or machine rather than as a Stream literally

Let’s take a look at the features of this machine:

• It has a portal for anything, and the machine doesn’t know when the portal will let anything in
• The machines in the middle can produce or process, which should take some time
• It has an outlet, there should be some product coming out of there, we don’t know when the product will come out of the outlet

The whole process, the timing is an uncertain factor, we can put something into the entrance of the machine at any time, and then the machine will be processed, but we don’t know how long it will be processed. So the exit had to be watched, waiting for things to come out of the machine. The whole process is viewed asynchronously

We convert the machine model into a Stream

• This big machine is the StreamController, which is one of the ways to create streams
• StreamController has an entry, called sink, that can use the Add method to put things in, and then it doesn’t care about them anymore
• StreamController has an outlet called stream, and when the machine finishes processing it will throw the product out of the outlet, but we don’t know when it will come out, so we need to listen to this outlet all the time. And when multiple items are put in, it doesn’t shuffle them, it’s first in, first out

支那

The Stream operation is also a tool provided in DART to handle asynchronous operations. Unlike a Future, it can receive the results of multiple asynchronous operations (whether they succeed or fail). You can pass the result data or error exception by triggering the success or failure event for many times. For example, we often see scenarios in the development process: multiple file reads and writes, network downloads may initiate multiple, etc. Next, we analyze the common operation of Stream according to a case:

void main(){
   Stream.fromFutures([loginFuture,getUserInfoFuture,saveUserInfoFuture])
         .listen((data){
          // A callback to the result returned by each task
         },onError:((err){
           // Failed callback for each task
         }),
         onDone : ((){
           // Listen for the completion of the callback during the execution of each task
         })
         )
         .onDone((){
           // All tasks completed callback
         });
}

Future loginFuture = Future<String> (() {// Call the login operation here
   login('admin'.'123456');
});
String login(String userName, String pwd){
  // Login operation
}
bool getUserInfo(int id){
  // Get user information
}
Future<String> getUserInfoFuture =Future((){
  getUserInfo(1);
});

Future saveUserInfoFuture = Future((){
   saveUserInfo("admin");
});

void saveUserInfo(String userInfo){

}
Copy the code

You can see that in listen method, we can perform fine callbacks for each task. Even after all tasks are completed, we also have callbacks such as Cancel, Pause, Resume, onError, onDone, which can perform fine callbacks for different stages of the execution of the whole group of tasks. In addition to using Stream as a fromFutures method, we often use Stream as a fromFuture method to handle a single Futrue/fromIterable collection. Of course, in addition to this regular Stream operation, DART provides two classes that specialize in manipulating/creating streams to implement the complex operation of Stream operations ** **

Method to get Stream

• By constructor
• Using StreamController
• IO Stream

Stream.fromIntreable([1.2.3]);
Copy the code

// Stream of any type
StreamController controller = StreamController();
controller.sink.add(123);
controller.sink.add("xyz");
controller.sink.add(()=>print("Output"));

// Create a stream that handles ints
StreamController<int> numController = StreamController();
numController.sink.add(123);
Copy the code

import 'dart:async';

void main(List<String> args) {
  StreamController controller = StreamController();
  controller.sink.add(123);
  controller.sink.add(() => print("Add a function"));

  // Listen for the exit of the stream and print the data when it comes out
  StreamSubscription subscription = controller.stream.listen((data) {
    if (data is Function) {
      data();
    } else {
      print("$data"); }}); }/* output 123 add a function */
Copy the code

Stream generation via async*

If we have a series of events to process, we might want to convert it to stream. You can use async -yield * to generate a Stream

Stream<int> countStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yieldi; }}Copy the code

When the loop exits, the Stream is done. We can combine with await for more profound experience

import 'dart:async';

void main() async {
  var stream = countStream(10);
  var sum = await sumStream(stream);
  print(sum);
}

Future<int> generateData(int data) async => data;

Stream<int> countStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield await generateData(i);
  }
}

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}
Copy the code

Click here to run the above sample code directly, and click here to see Promise and Yeild in JavaScript

Process the Stream with await for

In addition to listening, we can also use await for

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}
Copy the code

This code will receive a Stream, count all the events, and return the result. Await for can process each event as it comes. We know that a Stream is undefined when it receives events. When should we exit the await for loop? The answer is, when the Stream is finished or closed

(the stream object). The where ((event) {... })Copy the code

.take(4);
Copy the code

transform

If you need more control over the transformation, use the transform() method. It needs to work with StreamTransformer. Let’s look at a guessing game and I’ll explain it to you

import 'dart:async';

main(List<String> args) {
  
  StreamController<int> controller = StreamController<int> ();final transformer =
      StreamTransformer<int.String>.fromHandlers(handleData: (value, sink) {
    if (value == 100) {
      sink.add("You guessed it.");
    } else {
      sink.addError('Not yet. Try again.'); }}); controller.stream .transform(transformer) .listen((data) =>print(data), onError: (err) => print(err));

  controller.sink.add(23);
  //controller.sink.add(100);
}

// Output: not correctly guessed, try again
Copy the code

StreamTransformer<S,T> is the inspector for our stream. It receives the stream as it passes and processes it to return a new stream.

• S stands for the input type of the previous stream, so in this case we’re entering a number, so int.
• T stands for the input type of the transformed stream, and what we’re adding here is a String, so String.
• HandleData receives a value and creates a new stream and exposes sink, where we can convert the stream.
• We can also use addError to go in and tell you there’s a problem

Then we listen to the stream after the transform. When the converted event flows out, we print this event, which is the data we added to sink just now. OnError captures the errs we add.

“Single-subscription” Streams

A single subscription stream is allowed to have only one listener during the lifetime of the stream. It doesn’t generate events until there’s a listener, and it stops sending events when you cancel listening, even if you’re still sinking.add more events. Even after the first subscription is canceled, listening twice on a single subscription stream is not allowed. Single-subscription flows are often used to stream larger contiguous blocks of data, such as file I/O

import 'dart:async';

main(List<String> args) {
  StreamController controller = StreamController();
  controller.stream.listen((data) => print(data));
  controller.stream.listen((data) => print(data));
  controller.sink.add(123);
}


/ / output:Unhandled exception: Bad state: Stream has already been listened to. A single subscription stream cannot have multiple listeners0      _StreamController._subscribe (dart:async/stream_controller.dart:670:7)
#1      _ControllerStream._createSubscription (dart:async/stream_controller.dart:820:19)
#2      _StreamImpl.listen (dart:async/stream_impl.dart:474:9)
#3      main (file:///Users/Ken/Desktop/test/nam.dart:6:21)
#4      _startIsolate.<anonymous closure> (dart:isolate-patch/isolate_patch.dart:305:32)
#5      _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:174:12)
Copy the code

import 'dart:async';

main(List<String> args) {
  StreamController controller = StreamController();
	// Convert a single subscription stream to a broadcast stream
  Stream stream = controller.stream.asBroadcastStream();
  stream.listen((data) => print(data));
  stream.listen((data) => print(data));
  controller.sink.add(123);
}

// Output: 123 123
Copy the code