Dart Basics 5: Asynchronous support and exceptions

Dart Basic Directory

1.1 Mind mapping

1.2 Dart foundation will be explained in five parts:

one	Mainly explain keywords, variables, built-in types, operators, control flow statements
two	I’m going to focus on functions
three	Main Lecture class
four	Covers generics, libraries, and visibility
five	This section describes asynchronous support and exceptions

Dart thread model

Code execution in programming is usually divided into synchronous and asynchronous.

Synchronization: Simply put, synchronization is the top-down execution of code in the order in which it was written, and it’s the simplest form we’re most exposed to. However, the disadvantage of synchronizing code is obvious. If one or more lines of code are too time-consuming, they block, preventing subsequent code from being executed immediately.
Asynchrony: Asynchrony is designed to solve this problem by preventing some time-consuming code from being executed immediately on the current line of execution. The most common solution is to use multithreading, which is equivalent to opening up another line of execution and running time-consuming code on another line so that the two lines are side-by-side and time-consuming code cannot block code on the main line.

Multithreaded although good, but when a large number of simultaneous, there are still two major defects, one is open thread is consuming resources, thread open much is too much for the machine, the other is the thread lock problem, the need when multiple threads to Shared memory operation lock, lock competition will not only reduce the performance under complex conditions, can also cause a deadlock. Hence the emergence of the event-based asynchronous model.

The asynchronous model is simple said there is an event in a single thread loop and an event queue, events cycle continuously removed from the event queue to perform, the event is like a piece of code, every time when you meet the time-consuming event event loop will not stop and wait for the results, it will skip time-consuming events, continue to implement the following events. When the non-time-consuming events are complete, you can view the results of the time-consuming events. Thus, a time-consuming event does not block the entire event loop, giving subsequent events a chance to execute as well.

It’s easy to see that this event-based asynchronous model only worksI/O intensiveBecause I/O time-consuming operations tend to waste time waiting for the other party to send data or return results, this asynchronous model is often used for network server concurrency. If it isComputation-intensiveShould make use of the processor’s multi-core as far as possible to achieve parallel computing.

2.1 Single-threaded asynchronous operations

Single-threaded and asynchronous operations do not conflict:

Because one of our applications is idle most of the time, there is no unlimited interaction with the user.
For example, waiting for a user to click, the return of network request data, or the IO operation of a file read or write does not block our thread;
This is because IO, such as network requests and file reads and writes, can be called on a non-blocking basis;

To understand this, we need to understand the concept of blocking and non-blocking invocation in operating systems.

Blocking and non-blocking are concerned with the state of the program while it waits for the result of the call (message, return value).
Blocking call: The current thread is suspended until the result of the call is returned, and the calling thread will not resume execution until it gets the result of the call.
Non-blocking calls: After the call is executed, the current thread does not stop executing. It just needs to wait for a while to check if the result is returned.

Let’s use a real-life example:

You are hungry at noon and need to order a takeaway. Ordering a takeaway is our call and getting a takeaway is the result we have to wait for.
Blocking calls: ordering takeout and not doing anything is just waiting stupidly for your thread to stop doing anything else.
Non-blocking calls: You order a takeaway and continue to do something else: you continue to work, play a game, and your thread is not doing anything else, just checking occasionally to see if there’s a knock on the door or if the takeaway has arrived.

Many of the time-consuming operations in our development can be based on non-blocking calls like this:

For example, the network request itself uses Socket communication, and the Socket itself provides a SELECT model, which can be carried outWorks in a non-blocking manner;
For file read/write IO operations, we can use the event-based callback mechanism provided by the operating system.

None of these actions will block our single-thread execution, and our thread can continue to do other things while it waits: grab a cup of coffee, play a game, and wait for the actual response to be processed. So how does a single thread handle the results of network traffic, IO operations, and so on?

2.2 Event cycle

The single-threaded model basically maintains an Event Loop. What is the event loop?

In fact, the Event loop is not complicated, it is a series of events (including click events, IO events, network events) to be processed in an Event Queue.
Continually fetching events from the Event Queue and executing their corresponding blocks of code until the Event Queue is empty.

Let’s write pseudocode for an event loop:

// Here I use arrays to simulate queues, first in, first out
List eventQueue = []; 
var event;

// The event loop is always executed from the moment it starts
while (true) {
  if (eventQueue.length > 0) {
    // Retrieve an event
    event = eventQueue.removeAt(0);
    // Execute the eventevent(); }}Copy the code

When we have events, such as click events, IO events, network events, they are added to the eventLoop, and when it is found that the event queue is not empty, the event is fetched and executed. Here’s a piece of code to understand how the click event and the network request event are executed: in the RaisedButton onPressed function, a network request is sent when the button is pressed, and the callback in then is executed when the request succeeds.

RaisedButton(
  child: Text('Click me'),
  onPressed: () {
    final myFuture = http.get('https://example.com');
    myFuture.then((response) {
      if (response.statusCode == 200) {
        print('Success! '); }}); },)Copy the code

How is this code executed in an event loop?

When the user clicks, the onPressed callback is placed into an event loop and a network request is sent during execution.
After the network request is sent, the event loop is not blocked and is discarded when the onPressed function to execute is finished.
After the network request is successful, the callback function passed in THEN is executed, which is also an event. The event is put into the event loop for execution, and the event loop discards it after execution.

While onPressed is a little different than the callback in THEN, they both tell the event loop that I have a piece of code that needs to be executed, help me get it done.

2.3 Dart Event cycle Failure

Dart is an event-driven architecture based on a single-threaded execution model with a single event loop and two queues. Dart runs in a message loop mechanism in a single thread. There are two task queues: the microtask queue and the event queue. As shown in the following figure, the microtask queue has a higher execution priority than the event queue. Dart provides a call stack, but the event loop is supported by a single thread, so synchronization and locking are not required at all.

Dart event loop execution is shown in the figure above

As shown in the figure above, after the entry function main() is executed, the message loop mechanism starts.
Check whether the MicroTask queue is empty. If no, execute the MicroTask queue first.
After a MicroTask is executed, the Event queue is executed only when the MicroTask queue is empty.
After the Evnet queue has processed an event, go back to the second step to check if the MicroTask queue is empty.

Note: We can see that adding tasks to MicroTask can be executed as quickly as possible, but it is also important to note that when the Event loop is processing the MicroTask queue, the Event queue gets stuck and the application cannot process mouse clicks, I/O messages, and so on.

However, new microtasks and event tasks can also be inserted into the process of the event task execution. In this case, the whole thread execution is always in a loop and never exits, whereas the main thread execution of a Flutter never terminates. In the Dart, all the external events of tasks in the event queue, such as I/o, timer, click, and drawing events, etc., and micro task is usually based on the Dart, and micro tasks is very little, so, because the task queue priority, if the task is too much, total execution time is longer, the task of the event queue delay will be for a long time, The most intuitive representation for GUI applications is the comparison card, so it is important to ensure that the microtask queue is not too long. It is worth noting that we can use future.microTask (…) Inserts a task into the microtask queue.

2.4 Dart Execution model

When you start a Flutter (or any Dart application), a new thread process (” Isolate “in Dart) is created and started. This thread is the only one you need to focus on throughout the application, so Dart automatically:

Initialize two FIFO (first in, first out) queues (” MicroTask “and” Event “);
And when the method is done, execute main(),
Start the event loop.

Throughout the life of the thread, a single hidden process called an Event loop determines how and in what order your code executes (depending on microTasks and Event queues). An Event loop is an infinite loop (controlled by an internal clock), and within each clock cycle, If no other Dart code executes, do the following:

void eventLoop(){
    while (microTaskQueue.isNotEmpty){
        fetchFirstMicroTaskFromQueue();
        executeThisMicroTask();
        return;
    }
 
    if(eventQueue.isNotEmpty){ fetchFirstEventFromQueue(); executeThisEventRelatedCode(); }}Copy the code

As we have seen, the MicroTask queue takes precedence over the Event queue, so what are the two queues for? MicroTask queues are used for very short internal actions that need to be executed asynchronously, which need to be run after something else has completed and before the execution authority is returned to the Event queue. As an example of MicroTask, you can imagine that a resource must be released immediately after it is closed. Since the shutdown process may take some time to complete, you can write the code as follows:

MyResource myResource; .void closeAndRelease() {
    scheduleMicroTask(_dispose);
    _close();
}
 
void _close(){
    // The code runs synchronously
    // To close the resource. }void _dispose(){
    / / in the code
    / / _close () method
    // Execute after completion} Duplicate codeCopy the code

This is something you don’t have to use most of the time. For example, the scheduleMicroTask() method is referenced only 7 times in the entire Flutter source code. It is best to use the Event queue first. The Event queue applies to the following reference model

External events such as
- The I/O;
- Gestures;
- The drawing;
- The timer.
- Flow;
- …
futures

In fact, every time an external Event is triggered, the code to execute is referenced by the Event queue. Once no Micro Task is running, the Event loop considers the first item in the Event queue and executes it. It is worth noting that Future operations are also handled through the Event queue.

2.4.1 Scheduling tasks

There are two ways to add a task to the MicroTask queue:

Add using the scheduleMicrotask method.
Add using the Future object.

Note: The methods called below are defined in the DART: Async library.

import 'dart:async';

// My task queue
void myTask() {
  print("this is my task");
}

void main() {
  // 1. Use the scheduleMicrotask method
  scheduleMicrotask(myTask);

  // 2. Use the Future object to add
  Future.microtask(myTask);
}
Copy the code

Add the task to the Event queue using the Future object

import 'dart:async';

// My task
void myTask() {
  print("this is my task");
}

void main() {
// 1. Use the Future object to add
  Future(myTask);
}
Copy the code

Here’s an example:

import 'dart:async';

void main() {

  print('main Start');

   Future((){
    print('this is my task');
  });

   Future.microtask((){
    print('this is microtask');
  });

  print('main Stop');
}

// Result:
main Start
main Stop
this is microtask
this is my task

Copy the code

As you can see, the code does not run in the order we wrote it. Adding tasks to the queue does not mean they are executed immediately. They are executed asynchronously.

2.4.2 Delayed Tasks

If you need to delay the task, use the future.delayed method.

Future.delayed(new  Duration(seconds:1), () {print('task delayed');
});
Copy the code

Indicates that the task is added to the Event queue after the delay time expires. Note that this is not accurate, and your delayed task may not run on time in case there are time-consuming tasks ahead.

import 'dart:async';
import 'dart:io';

void main() {
  print("main start");

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

  Future(() {
    // The simulation takes 5 seconds
    sleep(Duration(seconds: 5));
    print("5s task");
  });

  print("main stop");
}

// Result:
main start
main stop
5s task
task delayed
Copy the code

It can be seen from the result that the delayed method was called first, but it obviously did not directly add the task to the Event queue. Instead, it waited 1 second to add the task. However, during this 1 second, the subsequent Future code directly added a time-consuming task to the Event queue. As a result, the delayed task written in the front can only be added to the time-consuming task after 1 second, and it can only be executed after the time-consuming task has been completed. This mechanism makes delayed tasks less reliable, and you can’t be sure how long the delayed tasks will be executed.

Asynchronous support

The Dart library contains many functions that return Future or Stream objects. These functions return immediately after setting up time-consuming tasks (such as I/O) and do not wait for time-consuming tasks to complete. Asynchronous programming with async and await keywords allows you to perform asynchronous operations as if you were writing synchronous code.

3.1 the Future

A Future is a task that executes asynchronously and completes (or fails) at some point in the Future. It is a proxy for Future results and does not return the return value of the invoked task. You can think of it as a Future is a client entrusted by you. You give him the task to be executed in the Future, and tell him whether the task type is time-consuming or non-time-consuming, and then classify the task into an event loop. When the task is completed, it will execute the callback method immediately to tell you that the task is completed. Or he’ll let you know as soon as all the tasks you’ve entrusted him with are completed. 支那

3.1.1 to create the Future

There are several ways to create a Future: • Future() • future.microtask () • future.sync () • future.value () • future.delayed () • future.error () The task will be executed immediately

import 'dart:async';

void main() {
  print("main start");

  Future.delayed(Duration(seconds: 1), () {
    print("delayed task");
  });
  
  Future.sync(() {
    print("sync task");
  });
  
  Future.microtask(() {
    print("microtask task");
  });
  
  Future(() {
    print("Future task");
  });

  print("main stop");
}

// Result:
main start
sync task
main stop
microtask task
Future task
delayed task
Copy the code

3.2 the Future state

In fact, during the entire implementation of a Future, we usually divide it into two states:

State 1: Uncompleted State

When performing an operation inside a Future, we call it an unfinished state.

State 2: Completed State

When the operation inside the Future completes, it usually returns a value or throws an exception.
In both cases, we call the Future a completed state.

Dart’s official website analyzes these two states, which are different from Promise’s three states.

3.3 Future common functions

Then () function, which will enter the THEN function after the task is executed and can obtain the returned result;
CatchError () function, where you can catch exceptions if the task fails;
The whenComplete() function, when the task is complete, enters here;
Wait (), which calls then() after multiple asynchronous tasks have completed;
Delayed () function, delayed the task execution;

When you instantiate a Future:

An instance of the Future is created and recorded in an internal array managed by Dart;
Code that needs to be executed by this Future is pushed directly to the Event queue;
The Future instance returns a state (= incomplete);
If the next synchronous code exists, execute it (non-future executable code)

As long as the Event loop gets it from the Event loop, the code referenced by the Future will execute like any other Event, and the THEN () or catchError() methods will be fired directly when the code will be executed and will complete (or fail).

To illustrate this point, let’s look at the following example:

void main() {
  
  print('Before the Future');
  
  Future(() {
    print('Running the Future');
  }).then((_) {
    print('Future is complete');
  });
  
  Future(() {
    print('Running');
  }).then((_) {
    print('complete');
  });
  
  print('After the Future');
}

// Run the result
Before the Future
After the Future
Running the Future
Future is complete
Running
complete
Copy the code

The execution process is as follows:

Output Before the Future;
Add Running the Future to the Event queue, add Running to the Event queue;
Output After the Future;
The event loop takes the code and executes it;
When the code executes, it looks for the THEN () statement and executes it;

Some very important things to keep in mind:

Futures are not executed in parallel, but in the same order that events are handled in an event loop. Look again at the example of creating a Future.

3.4 async and await

The async and await keywords have been added to Dart1.9. With these keywords, we can write asynchronous code more succinctly without calling future-related apis. They allow you to write asynchronous code as if you were writing synchronous code without using the Future interface. When you use the async keyword as the suffix for a method declaration, Dart will read it as:

The return value of this method is a Future;
It synchronizes the code executing the method up to the first await keyword, and then it suspends the rest of the method;
Once the Future referenced by the await keyword completes, the next line of code executes immediately.

This is important to understand because many developers think that await suspends the whole process until it completes, but this is not the case. They forgot how the cycle of events works… Let’s start with a simple example:

Future<String> doTask() async {
  return await Future(() {
    return "Ok";
  });
}

// Define a function for wrapping
test() async {
  var r = await doTask();
  print(r);
}

void main() {
  print("main start");
  test();
  print("main end");
}

// Output the result
main start
main end
Ok
Copy the code

If we want to output results sequentially, we need to add async and await keywords to the main method.

Future<String> doTask() async {
  return await Future(() {
    return "Ok";
  });
}

// Define a function for wrapping
test() async {
  var r = await doTask();
  print(r);
}

void main() async {
  print("main start");
  await test();
  print("main end");
}

// Output the result
main start
Ok
main end
Copy the code

To illustrate this better, let’s go through a more complex example and try to point out the results:

main() async {
  methodA();
  await methodB();
  await methodC('main');
  methodD();
}

methodA() {
  print('A');
}

methodB() async {
  print('B start');
  await methodC('B');
  print('B end');
}

methodC(String from) async {
  print('C start from $from');

  Future(() {
    print('C running Future from $from');
  }).then((_) {
    print('C end of Future from $from');
  });

  print('C end from $from');
}

methodD() {
  print('D');
}

Copy the code

The correct output sequence is: A B start C start from B C end from B B end C start from main C end from main D C running Future from B C end of Future From B C running Future from main C end of Future from main If you initially want to execute methodD() only at the end of all code in the sample code, you should write the code as follows:

main() async {
  methodA();
  await methodB();
  await methodC('main');
  methodD();
}

methodA() {
  print('A');
}

methodB() async {
  print('B start');
  await methodC('B');
  print('B end');
}

methodC(String from) async {
  print('C start from $from');

  await Future(() {
    print('C running Future from $from');
  }).then((_) {
    print('C end of Future from $from');
  });

  print('C end from $from');
}

methodD() {
  print('D');
}
Copy the code

The output sequence is: A B start C start from B C running Future from B C end of Future from B C end from B B end C start from main C running Future from main C end of Future from main C end from main D The fact that simply adding await in methodC() where the Future is defined changes the entire line. Also, keep in mind:

Async is not executed in parallel, but in accordance with the order in which events are processed in an event loop.

The last example I want to show you is as follows. What is the output of running method1 and method2? Could they be the same?

void method1() {
  List<String> myArray = <String> ['a'.'b'.'c'];
  print('before loop');
  myArray.forEach((String value) async {
    await delayedPrint(value);
  });
  print('end of loop');
}

void method2() async {
  List<String> myArray = <String> ['a'.'b'.'c'];
  print('before loop');
  for (int i = 0; i < myArray.length; i++) {
    await delayedPrint(myArray[i]);
  }
  print('end of loop');
}

Future<void> delayedPrint(String value) async {
  await Future.delayed(Duration(seconds: 1));
  print('delayedPrint: $value');
}
Copy the code

The answer:

method1()	method2()
1. before loop	1. before loop
2. end of loop	2. delayedPrint: a (after 1 second)
3. delayedPrint: a (after 1 second)	3. delayedPrint: b (1 second later)
4. delayedPrint: b (directly after)	4. delayedPrint: c (1 second later)
5. delayedPrint: c (directly after)	5. end of loop (right after)

Do you know the difference between their behavior and why? The answer is based on the fact that method1 uses the forEach() function to iterate over a set of numbers. With each iteration, it calls a new callback function marked async (and therefore a Future). Execute the callback until you encounter await, and then push the rest of the code to the Event queue. Once the iteration is complete, it executes the next statement: “print(‘ end of loop ‘)”. When the execution is complete, the event loop handles the three registered callbacks. With Method2, everything runs in the same “block” of code, so it can be executed sequentially line by line (in this case), and as you can see, even in seemingly simple code, we still need to keep in mind how event loops work.

3.2 the Stream

If a Future represents the result of a single calculation, a stream is a series of results, and you can listen to the stream for notifications about the results (data and errors) and the stream’s closure, and you can pause while listening to the stream or stop listening before the stream is complete. A Stream is what Dart calls an asynchronous data sequence, which is simply an asynchronous data queue. We know that queues are fifO, and streams are fifO. More figuratively, a Stream is like a conveyor belt. Can automatically transport items from one side to the other side. On the other side, as shown above, objects fall and disappear if no one grabs them.

But if we set a listener at the end, we can trigger the corresponding response behavior when the item reaches the end.

There are two types of streams in the Dart language, single-subscription, point-to-point, and broadcast streams.

3.2.1 Creating a Single Subscription Stream

The most common streams for a single subscription stream contain a series of events that are part of a larger whole. Events must be delivered in the correct order, and no events must be lost. This is the stream you get when you read a file or receive a Web request. Such streams can only be listened to once. Listening again later may mean missing the original event, and then the rest is meaningless. When you start listening, the data will be extracted and provided in blocks. The characteristic of single-subscription streams is that only one listener is allowed to exist, and even after the listener is cancelled, no listener is allowed to register again. There are nine constructors for creating a Stream, one of which is for constructing a broadcast Stream. Here we’ll focus on five of them for constructing a single-subscription Stream. periodic

void main(){
  test();
}
test() async{
  // Create a flow using periodic. The first argument is the interval time and the second argument is the callback function
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  // await for loop reads from the stream
  await for(var i in stream){
    print(i); }}// The value can be processed in the callback function, which returns the value directly
int callback(int value){
  return value;
}

// Output result:
0
1
2
3
4.Copy the code

This method starts at the integer 0 and generates a natural sequence at a specified interval, which is set to once every second. The callback function is used to process the generated integer before putting it into the Stream. It’s not handled here. It’s just returned. Note that this flow is infinite and does not have any constraint to stop it. You’ll see how to condition a stream later. fromFuture

void main(){
  test();
}

test() async{
  print("test start");
  Future<String> fut = Future((){
      return "async task";
  });
  // Create a Stream from the Future
  Stream<String> stream = Stream<String>.fromFuture(fut);
  await for(var s in stream){
    print(s);
  }
  print("test end");
}

// Output result:
test start
async task
test end

Copy the code

This method creates a Stream from a Future, puts it into the Stream when the Future execution is complete, and fetches the result of the task’s completion from the Stream. This usage is much like an asynchronous task queue. FromFutures creates a Stream from multiple futures. It puts a series of asynchronous tasks into the Stream, each Future executes in sequence, and then puts them into the Stream when they’re done

import  'dart:io';
void main() {
  test();
}

test() async {
  print("test start");
  Future<String> fut1 = Future(() {
    // The simulation takes 5 seconds
    sleep(Duration(seconds:5));
    return "async task1";
  });
  Future<String> fut2 = Future(() {
    return "async task2";
  });
  // Put multiple futures into a list and pass in the list
  Stream<String> stream = Stream<String>.fromFutures([fut1, fut2]);
  await for (var s in stream) {
    print(s);
  }
  print("test end");
}

// Output result:
test start
async task1
async task2
test end

Copy the code

FromIterable creates a Stream from a collection in much the same way as the example above

// Create a Stream from a list
Stream<int> stream = Stream<int>.fromIterable([1.2.3]);
Copy the code

Value this is a new method for Dart2.5 to create a Stream from a single value

test() async{
  Stream<bool> stream = Stream<bool>.value(false);
  // await for loop reads from the stream
  await for(var i in stream){
    print(i); }}Copy the code

3.2.3 Listening on a single subscription Stream

There are also three ways to listen to a Stream and get data from it. One is the await for loop we used above, which is officially recommended and looks more concise and friendly. The other two ways are to use the forEach method or listen method

Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  // With forEach, pass in a function to fetch and process data
  stream.forEach((int x){
    print(x);
  });
Copy the code

StreamSubscription Listen (void onData(T event), {Function onError, void onDone(), bool cancelOnError})

Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  stream.listen((x){
    print(x);
  });
Copy the code

Several optional parameters can also be used

test() async{
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  stream = stream.take(5);
  stream.listen(
    (x)=>print(x),
  onError: (e)=>print(e),
  onDone: ()=>print("onDone"));
}
Copy the code

OnError: Triggered when an Error occurs
OnDone: trigger when done
UnsubscribeOnError: Whether to cancel listening when the first Error is encountered. The default is false

3.2.4 Stream conversion

Take and takeWhile Stream take(int count) are used to limit the number of elements in the Stream

test() async{
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  // When three elements are added, the listener stops and the Stream closes
  stream = stream.take(3);
  await for(var i in stream){
    print(i); }}// Output result:
0
1
2
Copy the code

Stream.takeWhile(bool test(T element)) is similar to take except that it takes a function type and must return a bool

stream = stream.takeWhile((x){
    // Check the current element and cancel the listener if the condition is not met
    return x <= 3;
  });
Copy the code

The skip and skipWhile

test() async{
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  stream = stream.take(5);
  // Skip two elements from Stream
  stream = stream.skip(2);
  await for(var i in stream){
    print(i); }}// Output result:
2
3
4
Copy the code

Note that this method skips only when it gets an element from the Stream. The skipped element is still executed, and the elapsed time is still there. It just skips the result of execution. The Stream skipWhile(bool test(T element)) method is the same as the takeWhile method, passing in a function to judge the result and skipping the conditional. ToList Future toList() means to store all data in the Stream in a List

test() async{
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  stream = stream.take(5);
  List <int> data = await stream.toList(); 
  for(var i in data){ 
      print(i); }}Copy the code

The length property waits and gets the amount of all data in the stream

test() async{
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
  stream = stream.take(5);
  var len = await stream.length;
  print(len);
}
Copy the code

Map We can use the map method to traverse the stream data, similar to the listen effect, for example:

import 'dart:async';

void main() {
  StreamController controller = StreamController();
  controller.stream.map((data) => data += 1).listen((data) => print(data));
  controller.sink.add(123);
  controller.sink.add(456);
  controller.close();
}

// Output the result
124
457
Copy the code

Where method accepts an anonymous function as parameter. The parameter of the function is the data we add to sink. Data will be allowed to pass only when the return value of the function is true.

import 'dart:async';

void main() {
  StreamController<int> controller = StreamController();

  final whereStream = controller.stream.where((data) => data == 123);
  whereStream.listen((data) => print(data));

  controller.sink.add(123);
  controller.sink.add(456);
  controller.close();
  // output: 123
}

// Output the result
123
Copy the code

In the above code, the WHERE condition defines that the data must be the same as 123 to be listened to, so two pieces of data are added to the stream and only 123 is printed. Expand We can use expand to extend an existing stream. This method takes a method as an argument and returns data of type Iterable, for example:

import 'dart:async';

void main() {
  StreamController controller = StreamController();
  controller.stream
      .expand((data) => [data, data.toDouble()])
      .listen((data) => print(data));
  controller.sink.add(123);
  controller.sink.add(456);
  controller.close();
}
// Output the result
123
123.0
456
456.0
Copy the code

In the code above, expand’s callback returns a list of integers and their floating-point numbers each time, so the final print is four values. When we need to process more complex streams, we can use the transform method, which takes a StreamTransformer parameter, and each time we add data to sink, the data will be processed by the transform first. Such as:

import 'dart:async';

void main() {
  StreamController controller = StreamController();
  final transformer =
      StreamTransformer<int.String>.fromHandlers(handleData: (value, sink) {
    value == 123 ? sink.add('test success') : sink.addError('test error');
  });
  controller.stream
      .transform(transformer)
      .listen((data) => print(data), onError: (err) => print(err));
  controller.sink.add(123); // output: test success
  controller.sink.add(456); // output: test error
  controller.close();
}

Copy the code

In this code, each time we add data to sink, the StreamTransformer

fromHandlers method returns a new Stream, where S represents the data type we added, in this case int, T represents the type returned by handleData, and each handleData method returns a String. The handleData method has two parameters, value and sink. Value represents the data we added to sink, while sink is exposed for use. In the above example, We used sink’s add method and addError method.
,t>

3.2.5 StreamController

It is essentially a helper class for the Stream and can be used to control the entire Stream process.

import 'dart:async';
void main() {
  test();
}
test() async{
  / / create
  StreamController streamController = StreamController();
  // Add events
  streamController.add('element_1');
  streamController.addError("this is error");
  streamController.sink.add('element_2');
  streamController.stream.listen(
    print,
  onError: print,
  onDone: ()=>print("onDone"));
}
Copy the code

To use this class, you need to import ‘Dart :async’. The add method is the same as the sink.add method, which is used to place an element, and the addError method, which is used to generate an error, is used to listen for onError in the method. You can also pass in a specified stream in the StreamController

test() async{
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), (e)=>e);
  stream = stream.take(5);
  StreamController sc = StreamController();
  // Pass Stream in
  sc.addStream(stream);
  / / to monitor
  sc.stream.listen(
    print,
  onDone: ()=>print("onDone"));
}
Copy the code

Now look at the prototype StreamController, which has five optional parameters

factory StreamController(
      {void onListen(),
      void onPause(),
      void onResume(),
      onCancel(),
      bool sync: false})
Copy the code

Callback when onListen registers to listen
OnPause Callback when a stream is paused
OnResume Callback when the stream resumes
OnCancel Callback when the listener is canceled
Sync the duty to true said SynchronousStreamController synchronization controller, the default value is false, said an asynchronous controller

test() async{
  / / create
  StreamController sc = StreamController(
    onListen: ()=>print("onListen"),
    onPause: ()=>print("onPause"),
    onResume: ()=>print("onResume"),
    onCancel: ()=>print("onCancel"),
    sync:false
  );
  StreamSubscription ss = sc.stream.listen(print);
  sc.add('element_1');
  / / pause
  ss.pause();
  / / recovery
  ss.resume();
  / / cancel
  ss.cancel();
  / / close the flow
  sc.close();
}

// Output the result
onListen
onPause
onCancel
Copy the code

“Element_1” is not printed and “onResume” is not printed because the listener is cancelled and the stream is closed

3.2.6 radio flow

Broadcast stream Another type of stream is for a single message that can be processed at once. For example, this flow can be used for mouse events in a browser. You can start listening to such a stream at any time, and events are triggered as you listen. Multiple listeners can listen simultaneously, and you can listen again later after canceling a previous subscription. As follows, calling Listen twice in a normal single-subscription stream returns an error

test() async{
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), (e)=>e);
  stream = stream.take(5);
  stream.listen(print);
  stream.listen(print);
}
Copy the code

Unhandled exception:
Bad state: Stream has already been listened to.
Copy the code

A broadcast stream can allow multiple listeners, just like a broadcast stream, to access data from each listener. Note that if a listener is being added to the broadcast stream when the event is fired, the listener will not receive the event that is currently being fired. If you cancel listening, the listener immediately stops receiving events. There are two ways to create a broadcast Stream, one directly from the Stream and the other using the StreamController

test() async{
  // Call Stream's asBroadcastStream method
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), (e)=>e)
  .asBroadcastStream();
  stream = stream.take(5);
  stream.listen(print);
  stream.listen(print);
}
Copy the code

Using StreamController

test() async{
  // Create a broadcast stream
  StreamController sc = StreamController.broadcast();
  sc.stream.listen(print);
  sc.stream.listen(print);
  sc.add("event1");
  sc.add("event2");
}
Copy the code

3.2.7 StreamTransformer

This class enables us to perform data conversion on a Stream. These transformations are then pushed back into the flow so that all listeners registered with the flow can receive the constructor prototype

factory StreamTransformer.fromHandlers({
      void handleData(S data, EventSink<T> sink),
      void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
      void handleDone(EventSink<T> sink)
})
Copy the code

HandleData: Responds to any data event emitted from the stream. The parameters provided are the data from the emitted event, and EventSink, which represents an instance of the current stream where this transformation is taking place
HandleError: Responds to any error events emitted from the stream
HandleDone: Called when the stream has no more data to process. This is usually called when the close() method of a stream is called

void test() {
  StreamController sc = StreamController<int> ();// Create the StreamTransformer object
  StreamTransformer stf = StreamTransformer<int.double>.fromHandlers(
    handleData: (int data, EventSink sink) {
      // After manipulating the data, convert to type double
      sink.add((data * 2).toDouble());
    }, 
    handleError: (error, stacktrace, sink) {
      sink.addError('wrong: $error'); }, handleDone: (sink) { sink.close(); });// Call the stream's transform method, passing in the transform object
  Stream stream = sc.stream.transform(stf);
  stream.listen(print);
  // Add data, in this case of type int
  sc.add(1);
  sc.add(2); 
  sc.add(3); 
  
  // The handleDone callback is triggered
  // sc.close();
}

// Output result:
2.0
4.0
6.0
Copy the code

4. Multiple processes

Dart is a single-threaded asynchronous model, so can we run code in parallel in Dart? If you have a multi-core CPU, isn’t a single thread underusing the CPU? You need to rely on the Isolate.

4.1 Isolate

Multicore cpus are used in most computers, even on mobile platforms. In order to take advantage of multi-core performance, developers generally use shared memory data to ensure the correct execution of multiple threads. However, sharing data with multiple threads often causes many potential problems and causes code to run incorrectly. Dart is a language based on a single-threaded model. Dart also has its own process mechanism, ISOLATE. In Dart, all Dart code is run in _ quarantines instead of threads. Each quarantine has its own memory heap, ensuring that the state of each quarantine is not accessed by other quarantines. Dart is known to be single-threaded. This thread has its own memory space that it can access and the event loop that it needs to run. We can call this spatial system an Isolate. Such as UI rendering, user interaction, and so on. The Isolate does not share any resources **, so there is no lock contention. The Isolate relies on messaging to communicate, so there is no resource preemption. If you add time-consuming tasks to the event queue during development, you can still slow down the processing of the entire event loop or even block. So the asynchronous model based on event loops still has a lot of drawbacks, and that’s where we need Isolate, so how do we create Isolate?

Create Isolate Create a new Isolate spawnUri from the main Isolate

static Future<Isolate> spawnUri()
Copy the code

SpawnUri method has three required arguments:

The first is a Uri that specifies the path to a new Isolate code file,
The second is the argument List of type List,
The third is news feed.

Note that the code file used to run the new Isolate must contain a main function, which is the entry method for the new Isolate. The args parameter list in the main function corresponds to the second parameter in the spawnUri. If you do not need to upload parameters to the new Isolate, you can upload an empty List to this parameter. Code in main Isolate:

import 'dart:isolate'; 

void main() {
  print("main isolate start");
  create_isolate();
  print("main isolate stop");
}

// Create a new ISOLATE
create_isolate() async{
  ReceivePort rp = new ReceivePort();
  SendPort port1 = rp.sendPort;
  Isolate newIsolate = await Isolate.spawnUri(new Uri(path: "./other_task.dart"),"hello, isolate"."this is args"], port1);
  SendPort port2;
  rp.listen((message){
    print("main isolate message: $message");
    if (message[0] = =0){
      port2 = message[1];
    }else{ port2? .send([1."This message was sent by main ISOLATE."]); }});// You can kill the created ISOLATE by calling the following methods when appropriate
  // newIsolate.kill(priority: Isolate.immediate);
}
Copy the code

Dart file and code the new Isolate

import 'dart:isolate';
import  'dart:io';


void main(args, SendPort port1) {
  print("isolate_1 start");
  print("isolate_1 args: $args");

  ReceivePort receivePort = new ReceivePort();
  SendPort port2 = receivePort.sendPort;

  receivePort.listen((message){
    print("isolate_1 message: $message");
  });

  // Send SendPort created on the current ISOLATE to the main ISOLATE for communication
  port1.send([0, port2]);
  // The simulation takes 5 seconds
  sleep(Duration(seconds:5));
  port1.send([1."Isolate_1 task completed"]);

  print("isolate_1 stop");
}
Copy the code

Results of running main Isolate:

main isolate start
main isolate stop
isolate_1 start
isolate_1 args: [hello, isolate, this is args]
main isolate message: [0, SendPort]
isolate_1 stop
main isolate message: [1, isolATE_1 task complete.] Isolate_1 message: [1, this message was sent by main ISOLATE.]Copy the code

spawn

static Future<Isolate> spawn()
Copy the code

The spawn method takes two required arguments:

The first is the time consuming functions that need to run on the new Isolate
The second is the dynamic message, which is usually used to send SendPort objects of the main Isolate.

In addition to using spawnUri, the spawn method is more commonly used to create a new Isolate. We usually want to write the newly created Isolate code in the same file as the main Isolate code, and we don’t want to have two main functions. Instead, the specified time-consuming functions are run on the new Isolate, which facilitates code organization and code reuse. Spawn is used similar to spawnUri and is more concise, with a slight modification to the above example:

import 'dart:isolate'; 
import  'dart:io';

void main() {
  print("main isolate start");
  create_isolate();
  print("main isolate end");
}

// Create a new ISOLATE
create_isolate() async{
  ReceivePort rp = new ReceivePort();
  SendPort port1 = rp.sendPort;

  Isolate newIsolate = await Isolate.spawn(doWork, port1);

  SendPort port2;
  rp.listen((message){
    print("main isolate message: $message");
    if (message[0] = =0){
      port2 = message[1];
    }else{ port2? .send([1."This message was sent by main ISOLATE."]); }}); }// Process time-consuming tasks
void doWork(SendPort port1){
  print("new isolate start");
  ReceivePort rp2 = new ReceivePort();
  SendPort port2 = rp2.sendPort;

  rp2.listen((message){
    print("doWork message: $message");
  });

  // Send SendPort created in the new ISOLATE to the main ISOLATE for communication
  port1.send([0, port2]);
  // The simulation takes 5 seconds
  sleep(Duration(seconds:5));
  port1.send([1."DoWork task completed"]);

  print("new isolate end");
}
Copy the code

Running results:

main isolate start
main isolate end
new isolate start
main isolate message: [0, SendPort]
new isolate end
main isolate message: [1, doWork completed] doWork message: [1, this message was sent by main ISOLATE.]Copy the code

Spawn and spawnUri both create two processes, one for the main Isolate and the other for the new Isolate. Both processes are bound to the message channel in both directions. Even when the new Isolate completes its tasks, the process does not exit immediately. After using the Isolate created by yourself, call newisolate. kill(priority: ISOLate.immediate). Kill the Isolate immediately.

4.2 Isolate Communication Mechanism

In real development, we didn’t just start a new Isolate and not care about the results:

We need the new Isolate to calculate and inform the Main Isolate (the Isolate enabled by default) of the results.
The Isolate uses SendPort to communicate messages.
We can pass the Main Isolate’s send channel as a parameter when the Isolate is started and concurrency occurs;
Concurrency when execution is complete, this pipe can be used to send messages to the Main Isolate;

The entire message communication process is shown in the figure above:

The two Isolate communicate with each other through two pairs of Port objects, which consist of a ReceivePort object for receiving messages and a SendPort object for sending messages. The SendPort object is not created separately; it is already included in the ReceivePort object. Note that a pair of Port objects can send messages only in one direction, just like a water pipe. ReceivePort and SendPort are located at two ends of the water pipe. Water flows from SendPort to ReceivePort. Therefore, message communication between two ISOLates must require two such water pipes, which requires two pairs of Port objects.

Now that you understand how the Isolate message communication works, how does it work in the Dart code?

The ReceivePort object calls the LISTEN method, passing in a function that can listen for and execute incoming messages. The SendPort object calls the send() method to send the message. Send can be null,num, bool, double,String, List,Map, or a custom class.

4.3 Create THE Isolate on FLutter

Either way, creating an Isolate in Dart is a bit cumbersome, but Dart doesn’t officially offer more advanced encapsulation. However, if you want to create an Isolate in Flutter, there is a simpler API provided by the Flutter official to further encapsulate ReceivePort. If you just need to run some code to do some specific work, and you don’t need to interact with the Isolate once the work is done, there’s a very handy Helper called Compute. It mainly contains the following functions:

Make an Isolate,
Running a callback function on the ISOLATE and passing some data,
Returns the result of processing the callback function,
After the callback is executed, the Isolate is terminated.

Note: The code below is not the DART API, but the Flutter API, so it only runs in the Flutter project

import 'package:flutter/foundation.dart';
import  'dart:io';

// Create a new Isolate in which to run the task doWork
create_new_task() async{
  var str = "New Task";
  var result = await compute(doWork, str);
  print(result);
}


void doWork(String value){
  print("new isolate doWork start");
  // The simulation takes 5 seconds
  sleep(Duration(seconds:5));

  print("new isolate doWork end");
  return "complete:$value";
}
Copy the code

The first parameter is the function to be executed. The function must be a top-level function, not an instance method of the class, but a static method of the class. The second parameter is the dynamic message type, which can be an argument to the function being run. Need to pay attention to, the use of compute should import ‘package: flutter/foundation. The dart’ package.

4.4 Application Scenario of Isolate

Although THE Isolate is good, it can also be used in appropriate scenarios. Abuse of the Isolate is not recommended. You should use the event loop mechanism of Dart to process asynchronous tasks as much as possible, so as to make full use of the advantages of Dart. So when should you use Future and when should you use Isolate? One of the easiest ways to determine this is based on the average time of some tasks: if methods execute in milliseconds or tens of milliseconds, use the Future, and if a task requires hundreds of milliseconds or more, it is recommended to create a separate Isolate. In addition to this, there are also some reference scenarios • JSON decoding • encryption • image processing: such as clipping • Network requests: loading resources, images

Five, abnormal

The Dart single-threaded model is important to look at before introducing exception catching. Only with an understanding of the Dart code execution flow can we know where to go to catch exceptions. In Java and Objective-C (” OC “), if an exception occurs and is not caught, the program terminates, but this does not happen in Dart or JavaScript! Investigate its reason, this has relation with their operation mechanism. Java and OC are both multithreaded programming languages that cause the entire process to exit when any thread raises an exception that is not caught. Dart and JavaScript, however, are single-threaded models. In an event loop, when an exception occurs in one task and is not caught, the program does not exit. As a result, subsequent code in the current task is not executed, meaning that an exception in one task does not affect the execution of other tasks.

Dart code can throw and catch exceptions. Exceptions represent unknown error conditions. If the exception is not caught, it is thrown, causing the code that threw it to terminate. Unlike Java, all exceptions in Dart are non-checked exceptions. Methods do not declare exceptions they throw, nor do they require any to be caught.

Dart provides the Exception and Error types, as well as several subtypes. You can also define your own exception types. However, the Dart program can throw any non-NULL object, not just Exception and Error objects.

5.1 Throwing An Exception

Here is an example of throwing or throwing an exception:

throw FormatException('Expected at least 1 section');
Copy the code

We can also throw arbitrary objects:

throw 'Out of llamas! ';
Copy the code

Tip: High-quality production code typically implements an Exception throw of type Error or Exception.

Because a thrown exception is an expression, it can be used in => statements, and it can be thrown anywhere else where expressions are used:

void distanceTo(Point other) => throw UnimplementedError();
Copy the code

5.2 Catching synchronization Exceptions

Synchronized exceptions in Dart can be caught with a try/on/catch/finally code block exception, and the throw keyword can be used to explicitly throw an exception. Catching an exception prevents the exception from being passed on (unless it is rethrown). This exception can be handled by catching it:

try {
  breedMoreLlamas();
} on OutOfLlamasException {
  buyMoreLlamas();
}
Copy the code

By specifying multiple catch statements, you can handle code that may throw multiple types of exceptions. The first catch statement that matches the type of exception thrown handles the exception. If a catch statement does not specify a type, the statement can handle thrown objects of any type:

try {
  breedMoreLlamas();
} on OutOfLlamasException {
  // a special exception
  buyMoreLlamas();
} on Exception catch (e) {
  // Any other exceptions
  print('Unknown exception: $e');
} catch (e) {
  // No specified type, handle all exceptions
  print('Something really unknown: $e');
}
Copy the code

As shown in the code above, you can use both on and catch in a capture statement, or you can use them separately. Use on to specify the exception type and catch to catch the exception object. The catch() function can specify 1 or 2 arguments, the first of which is the exception object thrown and the second is the stack information (a StackTrace object).

try {
  / /...
} on Exception catch (e) {
  print('Exception details:\n $e');
} catch (e, s) {
  print('Exception details:\n $e');
  print('Stack trace:\n $s');
}
Copy the code

If you only need to partially handle an exception, you can use the keyword rethrow to rethrow it.

void misbehave() {
  try {
    dynamic foo = true;
    print(foo++); // Runtime error
  } catch (e) {
    print('misbehave() partially handled ${e.runtimeType}. ');
    rethrow; // Allow callers to see the exception.}}void main() {
  try {
    misbehave();
  } catch (e) {
    print('main() finished handling ${e.runtimeType}. '); }}Copy the code

A try-catch block cannot catch an asynchronous exception. Synchronous calls declared with the await keyword fall within the scope of synchronous exceptions and can be caught with a try-catch.

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

5.3 Catching Asynchronous Exceptions

Function error {bool test(Object error)} {bool test(Object error) {return true; If false is returned, the catch logic is not executed and the exception is not caught. By default, the catch logic is considered true. The function here is to fine processing exceptions, can be understood as the enhanced version of the on keyword in the synchronization exception, the input parameter is at most two respectively error and stack, both optional.

Future(() {
}).then((value){

}).catchError((error, stack) {

});

Future.delayed(Duration(seconds: 1)).then((e) => Future.error("xxx"));
Copy the code

5.4 the finally

The code in finally is executed whether or not an exception is thrown. If a catch does not match an exception, the exception is thrown again after the finally execution is complete:

try {
  breedMoreLlamas();
} finally {
  // Always clean up, even if an exception is thrown.
  cleanLlamaStalls();
}
Copy the code

After any matching catch execution is complete, finally is executed:

try {
  breedMoreLlamas();
} catch (e) {
  print('Error: $e'); // Handle the exception first.
} finally {
  cleanLlamaStalls(); // Then clean up.
}
Copy the code

For more details, see the Exceptions section.

References:

Dart Official Website
Dart Learning — Dart asynchronous programming
Understand the Dart asynchronism thoroughly
Flutter asynchronous programming: Future, Isolate and event loops
Dart language Stream details
The Flutter Dart review — exception catch and throw