Flutter 异步机制：Future
Flutter 基于 dart 语言，dart 本身是一个单线程模型，Future 也是基于单线程的异步机制，即基于事件循环实现的异步，与多线程实现的异步并不一样，比较类似于 Android 中的 Handler 机制，而所谓的异步，就是向事件循环中心发送一条消息，等待调度，在之后的某个时刻执行代码，但是这段代码还是在当前线程执行的，所以，如果使用 Future 执行耗时任务，它可能不会阻塞当前的 UI 流程，不过后续的一些 UI 操作还是会受到影响。 使用 Future 异步执行代码需要四个步骤：
- 创建任务
- 发送任务
- 执行任务
- 执行功能代码
创建任务
关于 Future 任务的创建，在应用层一般是这么写：
new Future(() {
doSomething();
});那么，从 Future 的构造函数开始，可以一窥 Future 任务创建的全过程。
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;
}result._complete(computation()) 即最后的执行功能代码的部分，参见第四小节，Timer.run 则会一步步创建任务。
static void run(void callback()) {
new Timer(Duration.zero, callback);
}
// third_party/sdk/sdk/lib/async/timer.dart
factory Timer(Duration duration, void callback()) {
if (Zone.current == Zone.root) {
// No need to bind the callback. We know that the root's timer will
// be invoked in the root zone.
return Zone.current.createTimer(duration, callback);
}
return Zone.current
.createTimer(duration, Zone.current.bindCallbackGuarded(callback));
}
// third_party/sdk/runtime/lib/timer_patch.dart
static Timer _createTimer(Duration duration, void callback()) {
// TODO(iposva): Remove _TimerFactory and use VMLibraryHooks exclusively.
if (_TimerFactory._factory == null) {
_TimerFactory._factory = VMLibraryHooks.timerFactory;
}
if (_TimerFactory._factory == null) {
throw new UnsupportedError("Timer interface not supported.");
}
int milliseconds = duration.inMilliseconds;
if (milliseconds < 0) milliseconds = 0;
return _TimerFactory._factory(milliseconds, (_) {
callback();
}, false);
}这里最后调用 _TimerFactory._factory 创建 Timer 实例，_TimerFactory._factory 来自于 VMLibraryHooks.timerFactory ，而 VMLibraryHooks.timerFactory 的设置时机可以一步步回溯至 InitDartInternal ：
// lib/ui/dart_runtime_hooks.cc
static void InitDartInternal(Dart_Handle builtin_library, bool is_ui_isolate) {
Dart_Handle print = GetFunction(builtin_library, "_getPrintClosure");
Dart_Handle internal_library = Dart_LookupLibrary(ToDart("dart:_internal"));
Dart_Handle result =
Dart_SetField(internal_library, ToDart("_printClosure"), print);
PropagateIfError(result);
if (is_ui_isolate) {
// Call |_setupHooks| to configure |VMLibraryHooks|.
Dart_Handle method_name = Dart_NewStringFromCString("_setupHooks");
result = Dart_Invoke(builtin_library, method_name, 0, NULL);
PropagateIfError(result);
}
Dart_Handle setup_hooks = Dart_NewStringFromCString("_setupHooks");
Dart_Handle io_lib = Dart_LookupLibrary(ToDart("dart:io"));
result = Dart_Invoke(io_lib, setup_hooks, 0, NULL);
PropagateIfError(result);
Dart_Handle isolate_lib = Dart_LookupLibrary(ToDart("dart:isolate"));
result = Dart_Invoke(isolate_lib, setup_hooks, 0, NULL);
PropagateIfError(result);
}
// third_party/sdk/runtime/lib/timer_impl.dart
@pragma("vm:entry-point", "call")
_setupHooks() {
VMLibraryHooks.timerFactory = _Timer._factory;
}
static Timer _factory(
int milliSeconds, void callback(Timer timer), bool repeating) {
if (repeating) {
return new _Timer.periodic(milliSeconds, callback);
}
return new _Timer(milliSeconds, callback);
}_Timer 是 Timer 的实现类，重复执行与不重复执行的 Timer 会调用不同的构造函数，但是二者殊途同归。
factory _Timer(int milliSeconds, void callback(Timer timer)) {
return _createTimer(callback, milliSeconds, false);
}
factory _Timer.periodic(int milliSeconds, void callback(Timer timer)) {
return _createTimer(callback, milliSeconds, true);
}
static Timer _createTimer(
void callback(Timer timer), int milliSeconds, bool repeating) {
// Negative timeouts are treated as if 0 timeout.
if (milliSeconds < 0) {
milliSeconds = 0;
}
// Add one because DateTime.now() is assumed to round down
// to nearest millisecond, not up, so that time + duration is before
// duration milliseconds from now. Using microsecond timers like
// Stopwatch allows detecting that the timer fires early.
int now = VMLibraryHooks.timerMillisecondClock();
int wakeupTime = (milliSeconds == 0) ? now : (now + 1 + milliSeconds);
_Timer timer =
new _Timer._internal(callback, wakeupTime, milliSeconds, repeating);
// Enqueue this newly created timer in the appropriate structure and
// notify if necessary.
timer._enqueue();
return timer;
}_internal 函数是 _Timer 的构造函数，_enqueue 函数将 Timer 放入队列等待执行：
// third_party/sdk/runtime/lib/timer_impl.dart
void _enqueue() {
if (_milliSeconds == 0) {
if (_firstZeroTimer == null) {
_lastZeroTimer = this;
_firstZeroTimer = this;
} else {
_lastZeroTimer._indexOrNext = this;
_lastZeroTimer = this;
}
// Every zero timer gets its own event.
_notifyZeroHandler();
} else {
_heap.add(this);
if (_heap.isFirst(this)) {
_notifyEventHandler();
}
}
}Timer 的延迟是否为 0 是一个分界线，它会将 Timer 分别插入 _lastZeroTimer 和 _heap 中，然后调用 _notifyZeroHandler 或 _notifyEventHandler 通知目标线程处理任务，接下来就是发送任务的过程了。
发送任务
以 _notifyZeroHandler 为例，
// third_party/sdk/runtime/lib/timer_impl.dart
static void _notifyZeroHandler() {
if (_sendPort == null) {
_createTimerHandler();
}
_sendPort.send(_ZERO_EVENT);
}首先，确保 _sendPort 的存在，然后，使用 _sendPort 发送一条 _ZERO_EVENT 消息。
// third_party/sdk/runtime/lib/timer_impl.dart
static void _createTimerHandler() {
assert(_receivePort == null);
assert(_sendPort == null);
_receivePort = new RawReceivePort(_handleMessage);
_sendPort = _receivePort.sendPort;
_scheduledWakeupTime = null;
}_receivePort 与 _sendPort 是一对用于通信的接口，首先调用 RawReceivePort 构造函数创建 _receivePort，并且传递了回调函数 _handleMessage ，然后从 _receivePort 中取出 _sendPort ，可见这个通信模型的重点就是 _receivePort 的构造过程。
// third_party/sdk/runtime/lib/isolate_patch.dart
@patch
factory RawReceivePort([Function handler]) {
_RawReceivePortImpl result = new _RawReceivePortImpl();
result.handler = handler;
return result;
}
factory _RawReceivePortImpl() native "RawReceivePortImpl_factory";
// third_party/sdk/runtime/lib/isolate.cc
DEFINE_NATIVE_ENTRY(RawReceivePortImpl_factory, 0, 1) {
ASSERT(
TypeArguments::CheckedHandle(zone, arguments->NativeArgAt(0)).IsNull());
Dart_Port port_id = PortMap::CreatePort(isolate->message_handler());
return ReceivePort::New(port_id, false /* not control port */);
}构造函数是一个 native 函数，在 native 中，首先调用 PortMap::CreatePort 创建出 Dart_Port ，然后调用 ReceivePort::New 创建 ReceivePort 实例，实例化之后，将回调函数 handler 保存到了 map 中，key 为 _get_id，这也是一个 native 函数。
创建 Dart_Port 的参数 isolate->message_handler() 的设置时机为 InitIsolate：
// third_party/sdk/runtime/vm/isolate.cc
Isolate* Isolate::InitIsolate(const char* name_prefix,
IsolateGroup* isolate_group,
const Dart_IsolateFlags& api_flags,
bool is_vm_isolate) {
// Setup the isolate message handler.
MessageHandler* handler = new IsolateMessageHandler(result);
ASSERT(handler != nullptr);
result->set_message_handler(handler);
}Dart_Port PortMap::CreatePort(MessageHandler* handler) {
ASSERT(handler != NULL);
MutexLocker ml(mutex_);
#if defined(DEBUG)
handler->CheckAccess();
#endif
Entry entry;
entry.port = AllocatePort();
entry.handler = handler;
entry.state = kNewPort;
// Search for the first unused slot. Make use of the knowledge that here is
// currently no port with this id in the port map.
ASSERT(FindPort(entry.port) < 0);
intptr_t index = entry.port % capacity_;
Entry cur = map_[index];
// Stop the search at the first found unused (free or deleted) slot.
while (cur.port != 0) {
index = (index + 1) % capacity_;
cur = map_[index];
}
// Insert the newly created port at the index.
ASSERT(index >= 0);
ASSERT(index < capacity_);
ASSERT(map_[index].port == 0);
ASSERT((map_[index].handler == NULL) ||
(map_[index].handler == deleted_entry_));
if (map_[index].handler == deleted_entry_) {
// Consuming a deleted entry.
deleted_--;
}
map_[index] = entry;
// Increment number of used slots and grow if necessary.
used_++;
MaintainInvariants();
if (FLAG_trace_isolates) {
OS::PrintErr(
"[+] Opening port: \n"
"\thandler: %s\n"
"\tport: %" Pd64 "\n",
handler->name(), entry.port);
}
return entry.port;
}在 CreatePort 中，先是调用 AllocatePort 创建一个端口（先随机生成一个，再判断这个端口没有被使用，就可以返回），然后构建出 Entry 并将其存在一个哈希表中。
RawReceivePort* ReceivePort::New(Dart_Port id,
bool is_control_port,
Heap::Space space) {
ASSERT(id != ILLEGAL_PORT);
Thread* thread = Thread::Current();
Zone* zone = thread->zone();
const SendPort& send_port =
SendPort::Handle(zone, SendPort::New(id, thread->isolate()->origin_id()));
ReceivePort& result = ReceivePort::Handle(zone);
{
RawObject* raw = Object::Allocate(ReceivePort::kClassId,
ReceivePort::InstanceSize(), space);
NoSafepointScope no_safepoint;
result ^= raw;
result.StorePointer(&result.raw_ptr()->send_port_, send_port.raw());
}
if (is_control_port) {
PortMap::SetPortState(id, PortMap::kControlPort);
} else {
PortMap::SetPortState(id, PortMap::kLivePort);
}
return result.raw();
}在这个函数里面，先是创建了 SendPort ，然后对 ReceivePort 进行了一些初始化操作，并将 ReceivePort 的 RawObject 返回，接着在 third_party/sdk/runtime/lib/timer_impl.dart#_createTimerHandler 中还要通过 _receivePort 取得 _sendPort，最终会调用 _get_sendport，这也是一个 native 函数：
DEFINE_NATIVE_ENTRY(RawReceivePortImpl_get_sendport, 0, 1) {
GET_NON_NULL_NATIVE_ARGUMENT(ReceivePort, port, arguments->NativeArgAt(0));
return port.send_port();
}GET_NON_NULL_NATIVE_ARGUMENT 取出 ReceivePort 实例，通过 arguments->NativeArgAt(0) ，但是 _get_sendport 并没有传递参数，也就是说这个参数就是 _RawReceivePortImpl 自身，也就是上面返回的 RawObject，然后返回它的 send_port 。
再回到 third_party/sdk/runtime/lib/timer_impl.dart#_notifyZeroHandler，_sendPort 调用 send 函数发送了一条消息，最终会调用 _sendInternal ，再转到 c++ 层：
// third_party/sdk/runtime/lib/isolate.cc
DEFINE_NATIVE_ENTRY(SendPortImpl_sendInternal_, 0, 2) {
GET_NON_NULL_NATIVE_ARGUMENT(SendPort, port, arguments->NativeArgAt(0));
// TODO(iposva): Allow for arbitrary messages to be sent.
GET_NON_NULL_NATIVE_ARGUMENT(Instance, obj, arguments->NativeArgAt(1));
const Dart_Port destination_port_id = port.Id();
const bool can_send_any_object = isolate->origin_id() == port.origin_id();
if (ApiObjectConverter::CanConvert(obj.raw())) {
PortMap::PostMessage(
Message::New(destination_port_id, obj.raw(), Message::kNormalPriority));
} else {
MessageWriter writer(can_send_any_object);
// TODO(turnidge): Throw an exception when the return value is false?
PortMap::PostMessage(writer.WriteMessage(obj, destination_port_id,
Message::kNormalPriority));
}
return Object::null();
}首先取出 SendPort 和 Instance 实例，然后创建出 Message 实例，最后调用 PortMap::PostMessage 发送消息。
// third_party/sdk/runtime/vm/port.cc
bool PortMap::PostMessage(std::unique_ptr<Message> message,
bool before_events) {
MutexLocker ml(mutex_);
intptr_t index = FindPort(message->dest_port());
if (index < 0) {
return false;
}
ASSERT(index >= 0);
ASSERT(index < capacity_);
MessageHandler* handler = map_[index].handler;
ASSERT(map_[index].port != 0);
ASSERT((handler != NULL) && (handler != deleted_entry_));
handler->PostMessage(std::move(message), before_events);
return true;
}根据 Message 的 dest_port 找到 index ，再从哈希表中取出 handler，这里的 handler 就是在初始化 Dart_Port 传入的 isolate->message_handler() ，也就是 IsolateMessageHandler 实例，不过 IsolateMessageHandler 并没有重写 PostMessage 函数。
// third_party/sdk/runtime/vm/message_handler.cc
void MessageHandler::PostMessage(std::unique_ptr<Message> message,
bool before_events) {
Message::Priority saved_priority;
{
saved_priority = message->priority();
if (message->IsOOB()) {
oob_queue_->Enqueue(std::move(message), before_events);
} else {
queue_->Enqueue(std::move(message), before_events);
}
if (paused_for_messages_) {
ml.Notify();
}
if (pool_ != nullptr && !task_running_) {
ASSERT(!delete_me_);
task_running_ = true;
const bool launched_successfully = pool_->Run<MessageHandlerTask>(this);
ASSERT(launched_successfully);
}
}
// Invoke any custom message notification.
MessageNotify(saved_priority);
}message 分为两种，oob 和非 oob，实际上就是优先级的区分：
// third_party/sdk/runtime/vm/message.h
typedef enum {
kNormalPriority = 0, // Deliver message when idle.
kOOBPriority = 1, // Deliver message asap.
// Iteration.
kFirstPriority = 0,
kNumPriorities = 2,
} Priority;不同优先级的 message 会被加入不同的队列，oob_queue_ 和 queue_，完了调用 MessageNotify ：
// third_party/sdk/runtime/vm/isolate.cc
void IsolateMessageHandler::MessageNotify(Message::Priority priority) {
if (priority >= Message::kOOBPriority) {
// Handle out of band messages even if the mutator thread is busy.
I->ScheduleInterrupts(Thread::kMessageInterrupt);
}
Dart_MessageNotifyCallback callback = I->message_notify_callback();
if (callback != nullptr) {
// Allow the embedder to handle message notification.
(*callback)(Api::CastIsolate(I));
}
}此处的 priority 是 message 的 priority ，当优先级为 kOOBPriority 时，会中断当前的任务去处理这个 message 。然后从 isolate 中取出 callback 执行，再看 callback 到底是谁。这要从 DartIsolate::Initialize 开始：
// runtime/dart_isolate.cc
bool DartIsolate::Initialize(Dart_Isolate dart_isolate, bool is_root_isolate) {
SetMessageHandlingTaskRunner(GetTaskRunners().GetUITaskRunner(),
is_root_isolate);
}
void DartIsolate::SetMessageHandlingTaskRunner(
fml::RefPtr<fml::TaskRunner> runner,
bool is_root_isolate) {
if (!is_root_isolate || !runner) {
return;
}
message_handling_task_runner_ = runner;
message_handler().Initialize(
[runner](std::function<void()> task) { runner->PostTask(task); });
}从这里就可以看出，后续的 callback 会在 message_handling_task_runner_ 中运行，但是它实际上还是 UITaskRunner ，所以说，Future 的异步是单线程的异步，然后再看：
// third_party/tonic/dart_message_handler.cc
void DartMessageHandler::Initialize(TaskDispatcher dispatcher) {1
// Only can be called once.
TONIC_CHECK(!task_dispatcher_ && dispatcher);
task_dispatcher_ = dispatcher;
Dart_SetMessageNotifyCallback(MessageNotifyCallback);
}
void DartMessageHandler::MessageNotifyCallback(Dart_Isolate dest_isolate) {
auto dart_state = DartState::From(dest_isolate);
TONIC_CHECK(dart_state);
dart_state->message_handler().OnMessage(dart_state);
}// third_party/sdk/runtime/vm/dart_api_impl.cc
DART_EXPORT void Dart_SetMessageNotifyCallback(
Dart_MessageNotifyCallback message_notify_callback) {
Isolate* isolate = Isolate::Current();
CHECK_ISOLATE(isolate);
{
NoSafepointScope no_safepoint_scope;
isolate->set_message_notify_callback(message_notify_callback);
}
if (message_notify_callback != nullptr && isolate->HasPendingMessages()) {
::Dart_ExitIsolate();
// If a new handler gets installed and there are pending messages in the
// queue (e.g. OOB messages for doing vm service work) we need to notify
// the newly registered callback, otherwise the embedder might never get
// notified about the pending messages.
message_notify_callback(Api::CastIsolate(isolate));
::Dart_EnterIsolate(Api::CastIsolate(isolate));
}
}从以上代码可以看出，message_notify_callback 就是 MessageNotifyCallback 函数，这个函数调用了 OnMessage ，而 task_dispatcher_ ，就是上面给出的 UITaskRunner 的 dispatcher 。
// third_party/tonic/dart_message_handler.cc
void DartMessageHandler::MessageNotifyCallback(Dart_Isolate dest_isolate) {
auto dart_state = DartState::From(dest_isolate);
TONIC_CHECK(dart_state);
dart_state->message_handler().OnMessage(dart_state);
}
void DartMessageHandler::OnMessage(DartState* dart_state) {
auto task_dispatcher_ = dart_state->message_handler().task_dispatcher_;
// Schedule a task to run on the message loop thread.
auto weak_dart_state = dart_state->GetWeakPtr();
task_dispatcher_([weak_dart_state]() {
if (auto dart_state = weak_dart_state.lock()) {
dart_state->message_handler().OnHandleMessage(dart_state.get());
}
});
}从 OnHandleMessage 开始，剩下的代码开始在 task_dispatcher_ 中执行，也可以说从这里开始，开始了接收任务阶段。
接收任务
void DartMessageHandler::OnHandleMessage(DartState* dart_state) {
if (isolate_had_fatal_error_) {
// Don't handle any more messages.
return;
}
DartIsolateScope scope(dart_state->isolate());
DartApiScope dart_api_scope;
Dart_Handle result = Dart_Null();
bool error = false;
// On the first message, check if we should pause on isolate start.
if (!handled_first_message()) {
set_handled_first_message(true);
if (Dart_ShouldPauseOnStart()) {
// Mark that we are paused on isolate start.
Dart_SetPausedOnStart(true);
}
}
if (Dart_IsPausedOnStart()) {
} else if (Dart_IsPausedOnExit()) {
} else {
// We are processing messages normally.
result = Dart_HandleMessage();
// If the Dart program has set a return code, then it is intending to shut
// down by way of a fatal error, and so there is no need to emit a log
// message.
if (dart_state->has_set_return_code() && Dart_IsError(result) &&
Dart_IsFatalError(result)) {
error = true;
} else {
error = LogIfError(result);
}
dart_state->MessageEpilogue(result);
if (!Dart_CurrentIsolate()) {
isolate_exited_ = true;
return;
}
}
}正常情况下会调用 Dart_HandleMessage：
// third_party/sdk/runtime/vm/dart_api_impl.cc
DART_EXPORT Dart_Handle Dart_HandleMessage() {
Thread* T = Thread::Current();
Isolate* I = T->isolate();
CHECK_API_SCOPE(T);
CHECK_CALLBACK_STATE(T);
API_TIMELINE_BEGIN_END_BASIC(T);
TransitionNativeToVM transition(T);
if (I->message_handler()->HandleNextMessage() != MessageHandler::kOK) {
return Api::NewHandle(T, T->StealStickyError());
}
return Api::Success();
}
// third_party/sdk/runtime/vm/message_handler.cc
MessageHandler::MessageStatus MessageHandler::HandleNextMessage() {
// We can only call HandleNextMessage when this handler is not
// assigned to a thread pool.
MonitorLocker ml(&monitor_);
ASSERT(pool_ == NULL);
ASSERT(!delete_me_);
#if defined(DEBUG)
CheckAccess();
#endif
return HandleMessages(&ml, true, false);
}
MessageHandler::MessageStatus MessageHandler::HandleMessages(
MonitorLocker* ml,
bool allow_normal_messages,
bool allow_multiple_normal_messages) {
ASSERT(monitor_.IsOwnedByCurrentThread());
// Scheduling of the mutator thread during the isolate start can cause this
// thread to safepoint.
// We want to avoid holding the message handler monitor during the safepoint
// operation to avoid possible deadlocks, which can occur if other threads are
// sending messages to this message handler.
//
// If isolate() returns nullptr [StartIsolateScope] does nothing.
ml->Exit();
StartIsolateScope start_isolate(isolate());
ml->Enter();
MessageStatus max_status = kOK;
Message::Priority min_priority =
((allow_normal_messages && !paused()) ? Message::kNormalPriority
: Message::kOOBPriority);
std::unique_ptr<Message> message = DequeueMessage(min_priority);
while (message != nullptr) {
intptr_t message_len = message->Size();
if (FLAG_trace_isolates) {
OS::PrintErr(
"[<] Handling message:\n"
"\tlen: %" Pd
"\n"
"\thandler: %s\n"
"\tport: %" Pd64 "\n",
message_len, name(), message->dest_port());
}
// Release the monitor_ temporarily while we handle the message.
// The monitor was acquired in MessageHandler::TaskCallback().
ml->Exit();
Message::Priority saved_priority = message->priority();
Dart_Port saved_dest_port = message->dest_port();
MessageStatus status = HandleMessage(std::move(message));
if (status > max_status) {
max_status = status;
}
ml->Enter();
if (FLAG_trace_isolates) {
OS::PrintErr(
"[.] Message handled (%s):\n"
"\tlen: %" Pd
"\n"
"\thandler: %s\n"
"\tport: %" Pd64 "\n",
MessageStatusString(status), message_len, name(), saved_dest_port);
}
// If we are shutting down, do not process any more messages.
if (status == kShutdown) {
ClearOOBQueue();
break;
}
// Remember time since the last message. Don't consider OOB messages so
// using Observatory doesn't trigger additional idle tasks.
if ((FLAG_idle_timeout_micros != 0) &&
(saved_priority == Message::kNormalPriority)) {
idle_start_time_ = OS::GetCurrentMonotonicMicros();
}
// Some callers want to process only one normal message and then quit. At
// the same time it is OK to process multiple OOB messages.
if ((saved_priority == Message::kNormalPriority) &&
!allow_multiple_normal_messages) {
// We processed one normal message. Allow no more.
allow_normal_messages = false;
}
// Reevaluate the minimum allowable priority. The paused state
// may have changed as part of handling the message. We may also
// have encountered an error during message processing.
//
// Even if we encounter an error, we still process pending OOB
// messages so that we don't lose the message notification.
min_priority = (((max_status == kOK) && allow_normal_messages && !paused())
? Message::kNormalPriority
: Message::kOOBPriority);
message = DequeueMessage(min_priority);
}
return max_status;
}直到 MessageHandler::HandleMessages 为止，这里又是一个 while 循环，不断调用 DequeueMessage 取出 message ，直到所有的 message 执行完毕，单个 message 的处理，则是调用 HandleMessage ，
在 HandleMessage 中首先做的是获取 msg_handler ，调用的是 DartLibraryCalls::LookupHandler ，
RawObject* DartLibraryCalls::LookupHandler(Dart_Port port_id) {
Thread* thread = Thread::Current();
Zone* zone = thread->zone();
Function& function = Function::Handle(
zone, thread->isolate()->object_store()->lookup_port_handler());
const int kTypeArgsLen = 0;
const int kNumArguments = 1;
if (function.IsNull()) {
Library& isolate_lib = Library::Handle(zone, Library::IsolateLibrary());
ASSERT(!isolate_lib.IsNull());
const String& class_name = String::Handle(
zone, isolate_lib.PrivateName(Symbols::_RawReceivePortImpl()));
const String& function_name = String::Handle(
zone, isolate_lib.PrivateName(Symbols::_lookupHandler()));
function = Resolver::ResolveStatic(isolate_lib, class_name, function_name,
kTypeArgsLen, kNumArguments,
Object::empty_array());
ASSERT(!function.IsNull());
thread->isolate()->object_store()->set_lookup_port_handler(function);
}
const Array& args = Array::Handle(zone, Array::New(kNumArguments));
args.SetAt(0, Integer::Handle(zone, Integer::New(port_id)));
const Object& result =
Object::Handle(zone, DartEntry::InvokeFunction(function, args));
return result.raw();
}可以看出，这就是一个典型的 c++ 调用 dart 的流程，先找到 function ，然后构建参数，最后 DartEntry::InvokeFunction 调用这个函数。从 8～20 行得知，这是 _RawReceivePortImpl 的 _lookupHandler 函数：
@pragma("vm:entry-point", "call")
static _lookupHandler(int id) {
var result = _handlerMap[id];
return result;
}根据 id 从 _handlerMap 中找到一个值返回，这个值，其实就是会调函数，设置时机如下：
void set handler(Function value) {
_handlerMap[this._get_id()] = value;
}这就是在初始化 _RawReceivePortImpl 之后调用的，_get_id 返回的也正是 dest_id 。然后，HandleMessage 中对 message 分为三种情况进行处理：
- message 优先级为 oob
- message dest_port 为 kIllegalPort
- 正常情况
正常的处理如下：
const Object& result =
Object::Handle(zone, DartLibraryCalls::HandleMessage(msg_handler, msg));
if (result.IsError()) {
status = ProcessUnhandledException(Error::Cast(result));
} else {
ASSERT(result.IsNull());
}DartLibraryCalls::HandleMessage：
// third_party/sdk/runtime/vm/dart_entry.cc
RawObject* DartLibraryCalls::HandleMessage(const Object& handler,
const Instance& message) {
Thread* thread = Thread::Current();
Zone* zone = thread->zone();
Isolate* isolate = thread->isolate();
Function& function = Function::Handle(
zone, isolate->object_store()->handle_message_function());
const int kTypeArgsLen = 0;
const int kNumArguments = 2;
if (function.IsNull()) {
Library& isolate_lib = Library::Handle(zone, Library::IsolateLibrary());
ASSERT(!isolate_lib.IsNull());
const String& class_name = String::Handle(
zone, isolate_lib.PrivateName(Symbols::_RawReceivePortImpl()));
const String& function_name = String::Handle(
zone, isolate_lib.PrivateName(Symbols::_handleMessage()));
function = Resolver::ResolveStatic(isolate_lib, class_name, function_name,
kTypeArgsLen, kNumArguments,
Object::empty_array());
ASSERT(!function.IsNull());
isolate->object_store()->set_handle_message_function(function);
}
const Array& args = Array::Handle(zone, Array::New(kNumArguments));
args.SetAt(0, handler);
args.SetAt(1, message);
#if !defined(PRODUCT)
if (isolate->debugger()->IsStepping()) {
// If the isolate is being debugged and the debugger was stepping
// through code, enable single stepping so debugger will stop
// at the first location the user is interested in.
isolate->debugger()->SetResumeAction(Debugger::kStepInto);
}
#endif
const Object& result =
Object::Handle(zone, DartEntry::InvokeFunction(function, args));
ASSERT(result.IsNull() || result.IsError());
return result.raw();
}这又是从 c++ 中调用 dart 函数，从代码可知，调用的是 _RawReceivePortImpl 的 _handleMessage 函数：
@pragma("vm:entry-point", "call")
static void _handleMessage(Function handler, var message) {
// TODO(floitsch): this relies on the fact that any exception aborts the
// VM. Once we have non-fatal global exceptions we need to catch errors
// so that we can run the immediate callbacks.
handler(message);
_runPendingImmediateCallback();
}以 message 为参数，调用 handler 函数，即最初传进来的 _Timer 的 _handleMessage 函数：
// third_party/sdk/runtime/lib/timer_impl.dart
static void _handleMessage(msg) {
var pendingTimers;
if (msg == _ZERO_EVENT) {
pendingTimers = _queueFromZeroEvent();
assert(pendingTimers.length > 0);
} else {
assert(msg == _TIMEOUT_EVENT);
_scheduledWakeupTime = null; // Consumed the last scheduled wakeup now.
pendingTimers = _queueFromTimeoutEvent();
}
_runTimers(pendingTimers);
// Notify the event handler or shutdown the port if no more pending
// timers are present.
_notifyEventHandler();
}这里根据 msg 的不同从不同的队列中取 pendingTimers ，二者分别实现如下：
static List _queueFromZeroEvent() {
var pendingTimers = new List();
assert(_firstZeroTimer != null);
// Collect pending timers from the timer heap that have an expiration prior
// to the currently notified zero timer.
var timer;
while (!_heap.isEmpty && (_heap.first._compareTo(_firstZeroTimer) < 0)) {
timer = _heap.removeFirst();
pendingTimers.add(timer);
}
// Append the first zero timer to the pending timers.
timer = _firstZeroTimer;
_firstZeroTimer = timer._indexOrNext;
timer._indexOrNext = null;
pendingTimers.add(timer);
return pendingTimers;
}
static List _queueFromTimeoutEvent() {
var pendingTimers = new List();
if (_firstZeroTimer != null) {
// Collect pending timers from the timer heap that have an expiration
// prior to the next zero timer.
// By definition the first zero timer has been scheduled before the
// current time, meaning all timers which are "less than" the first zero
// timer are expired. The first zero timer will be dispatched when its
// corresponding message is delivered.
var timer;
while (!_heap.isEmpty && (_heap.first._compareTo(_firstZeroTimer) < 0)) {
timer = _heap.removeFirst();
pendingTimers.add(timer);
}
} else {
// Collect pending timers from the timer heap which have expired at this
// time.
var currentTime = VMLibraryHooks.timerMillisecondClock();
var timer;
while (!_heap.isEmpty && (_heap.first._wakeupTime <= currentTime)) {
timer = _heap.removeFirst();
pendingTimers.add(timer);
}
}
return pendingTimers;
}当 msg 为 _ZERO_EVENT 时，会取出一个 _firstZeroTimer 队列中的任务和 n 个 _heap 队列中达到执行时间的任务，而当 msg 不为 _ZERO_EVENT 时，则会取出 n 个执行时间先于 _firstZeroTimer 第一个任务的的任务，或者是执行时间先于当前时间的任务。取完之后则是调用 _runTimers 执行任务。
static void _runTimers(List pendingTimers) {
// If there are no pending timers currently reset the id space before we
// have a chance to enqueue new timers.
if (_heap.isEmpty && (_firstZeroTimer == null)) {
_idCount = 0;
}
// Fast exit if no pending timers.
if (pendingTimers.length == 0) {
return;
}
// Trigger all of the pending timers. New timers added as part of the
// callbacks will be enqueued now and notified in the next spin at the
// earliest.
_handlingCallbacks = true;
var i = 0;
try {
for (; i < pendingTimers.length; i++) {
// Next pending timer.
var timer = pendingTimers[i];
timer._indexOrNext = null;
// One of the timers in the pending_timers list can cancel
// one of the later timers which will set the callback to
// null. Or the pending zero timer has been canceled earlier.
if (timer._callback != null) {
var callback = timer._callback;
if (!timer._repeating) {
// Mark timer as inactive.
timer._callback = null;
} else if (timer._milliSeconds > 0) {
var ms = timer._milliSeconds;
int overdue =
VMLibraryHooks.timerMillisecondClock() - timer._wakeupTime;
if (overdue > ms) {
int missedTicks = overdue ~/ ms;
timer._wakeupTime += missedTicks * ms;
timer._tick += missedTicks;
}
}
timer._tick += 1;
callback(timer);
// Re-insert repeating timer if not canceled.
if (timer._repeating && (timer._callback != null)) {
timer._advanceWakeupTime();
timer._enqueue();
}
// Execute pending micro tasks.
var immediateCallback = _removePendingImmediateCallback();
if (immediateCallback != null) {
immediateCallback();
}
}
}
} finally {
_handlingCallbacks = false;
// Re-queue timers we didn't get to.
for (i++; i < pendingTimers.length; i++) {
var timer = pendingTimers[i];
timer._enqueue();
}
_notifyEventHandler();
}
}基本逻辑就是一个 for 循环，对于每一个 timer，执行其 callback 函数，一步步回溯回去，就会调用到 result._complete(computation()) 函数，从此处开始，就到了上层逻辑代码。
执行功能代码
接收到的最终的任务就是一个 callback，参见：
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;
}_complete 函数用于处理 computation 函数执行的结果，分为三种类型判断：
void _complete(FutureOr<T> value) {
assert(!_isComplete);
if (value is Future<T>) {
if (value is _Future<T>) {
_chainCoreFuture(value, this);
} else {
_chainForeignFuture(value, this);
}
} else {
_FutureListener listeners = _removeListeners();
_setValue(value);
_propagateToListeners(this, listeners);
}
}当返回结果还是一个 Future 类型时，和当返回结果不是 Future 对象时有这两种不同的处理逻辑，后者，也就是 else 语句中的逻辑，可以看到就是将返回值传递给 listener ，然后调用 _propagateToListeners 将结果分发给 listener 的 listeners，但前者，if 语句中的逻辑，告诉我们当上一个 Future 返回出一个 Future 时，不能直接将这个 Future 当作 listener 的 value 然后通知 listener ，这涉及到 Future 使用过程中的一个规则：Listener 的监听结果不能是一个 Future ，所以就有了如下这种代码：
test4() {
print("start");
Future(() {
print("return future");
return Future(() {
print("return test4");
return "test4";
});
}).then((FutureOr<String> s) {
if (s is Future<String>) {
s.then((String ss) {
print(ss + "_");
});
} else {
print(s);
}
});
print("end");
}这段代码最终打印的是 test4 而不是 test4_ ，而这种情况出现的原因，就在于 _complete 函数中对 Future 对象和非 Future 对象的两种不同的处理。对于非 Future 对象，将返回值通过 _setValue 传递给 result ，再分发给其 listener ，这是一个比较常规的过程。而对于 Future 对象也有两种不同的处理，当返回值是一个 _Future 对象时，调用 _chainCoreFuture ，否则调用 _chainForeignFuture ，这两个函数功能上大同小异，只不过对于 Flutter 默认实现的 _Future，有一些更“本土化”的处理，而对于其他 Future 的实现，则只能使用 Future 带有的函数来实现同样的功能，所谓的处理，就是 Future 不能作为 listener 的返回值出现，具体看代码：
static void _chainForeignFuture(Future source, _Future target) {
assert(!target._isComplete);
assert(source is! _Future);
// Mark the target as chained (and as such half-completed).
target._setPendingComplete();
try {
source.then((value) {
assert(target._isPendingComplete);
// The "value" may be another future if the foreign future
// implementation is mis-behaving,
// so use _complete instead of _completeWithValue.
target._clearPendingComplete(); // Clear this first, it's set again.
target._complete(value);
},
// TODO(floitsch): eventually we would like to make this non-optional
// and dependent on the listeners of the target future. If none of
// the target future's listeners want to have the stack trace we don't
// need a trace.
onError: (error, [StackTrace stackTrace]) {
assert(target._isPendingComplete);
target._completeError(error, stackTrace);
});
} catch (e, s) {
// This only happens if the `then` call threw synchronously when given
// valid arguments.
// That requires a non-conforming implementation of the Future interface,
// which should, hopefully, never happen.
scheduleMicrotask(() {
target._completeError(e, s);
});
}
}source 是上一个 Future 的返回值，target 是 listener 对应的 Future ，它没有将 source 设置为 target 的 value，而是继续调用 source 的 then 函数，得到 source 的返回值后再调用 target 的 _complete 函数处理返回值，这里便会陷入循环，如果 source 嵌套返回了多少个 Future ，这里就需要调用多少次，总之，直到 source 返回了一个常规的值，才会结束这种循环，这就是 _chainForeignFuture 中所做的处理，只使用到了 Future 的 then 函数，看起来也比较容易理解，不过 _chainCoreFuture 看着就没有这么和蔼了。
static void _chainCoreFuture(_Future source, _Future target) {
assert(target._mayAddListener); // Not completed, not already chained.
while (source._isChained) {
source = source._chainSource;
}
if (source._isComplete) {
_FutureListener listeners = target._removeListeners();
target._cloneResult(source);
_propagateToListeners(target, listeners);
} else {
_FutureListener listeners = target._resultOrListeners;
target._setChained(source);
source._prependListeners(listeners);
}
}首先判断 source 是否处于 Future 链中，如果是，则找到链尾的 Future ，然后将这个 Future 作为 source ，如果 source 已经完成了，把 source 的返回值复制过来，然后分发给 listeners，如果没有，则加入这个 Future 链，并将自己的 listeners 都移植到 source 身上。 那么 Future 链是什么？链尾的 source 又是什么？为什么 target 的 listeners 能够直接交给 source ？欢迎来到。。。。 先看几个示例代码：
test5() {
Future<String> f = Future.delayed(Duration(milliseconds: 100), () {
print("return test5");
return "test5";
});
Future(() {
print("return f");
return f;
}).then((String s) => print(s));
}
test6() {
Future<String> f = Future.delayed(Duration(milliseconds: 100), () {
print("return test6");
return "test6";
});
Future<String> ff = Future(() {
print("return f");
return f;
});
Future<String> fff = Future(() {
print("return ff");
return ff;
});
Future(() {
print("return fff");
return fff;
}).then((String s) => print(s));
}先看 test5 ，首先声明了一个 Future f，延迟 100 ms，接着又定义了一个 Future 会返回 f，它有一个 listener 需要接受 String 参数。当第二个 Future 返回 f 时 f 还未执行完毕，于是此时执行的代码段就是：
_FutureListener listeners = target._resultOrListeners;
target._setChained(source);
source._prependListeners(listeners);即 Future 加入了 Future 链，并将 listner 移交给了 f，所以此时实际上的逻辑应该是这样的：
test5() {
Future<String> f = Future.delayed(Duration(milliseconds: 100), () {
print("return test5");
return "test5";
});
f.then((String s) => print(s));
}于是一个双层嵌套的 Future ，变成了一个没有嵌套的 Future，当 f 执行完了之后，就会通知 listner 打印 s。 而对于 test6 ，这是一个多层嵌套，ff、fff 以及最后一个 Future ，返回的都是 Future ，那么当 ff 执行完时，它返回了一个尚未执行完的 f，所以它会加入到 Future 链，此时链为 ff -> f，然后 fff 返回了 ff，而此时 ff 已经在 Future 链中，所以 fff 也会加入 Future 链，此时 fff -> ff -> f，然后就是 Future -> fff -> ff -> f，并且此时 Future 的 listener 挂在了 f 身上，于是也可以化简为：
test6() {
Future<String> f = Future.delayed(Duration(milliseconds: 100), () {
print("return test6");
return "test6";
});
f.then((String s) => print(s));
}最终，当 f 执行完了之后，会进入 _complete 函数执行如下代码：
_FutureListener listeners = _removeListeners();
_setValue(value);
_propagateToListeners(this, listeners);此时的 listeners ，便是包含了 ff，fff 和 Future 这三个 Future 的 listener 集合，虽然如此，但此刻它们都已经移植到了 f 身上，并且这个移植是无缝的，从根本上来说，他们的 listener 实际上监听的也正是 f 的返回值。 看到这里，上面几个问题应该有了解答，Future 链是一串 Future ，但是它们中只有一个 Future 返回了真实的数据，也就是链尾的那一个，前面的 Future 的 listeners 则是直接或间接地依赖着链尾 Future 的返回结果，所以它们的 listeners 也都直接移植到了链尾 Future 上，这个链尾的 Future ，就是 _chainCoreFuture 中的 source，而对于其它那些没有返回真正数据的 Future ，是不会调用它们的 _propagateToListeners 函数的。 _propagateToListeners 负责将返回值传递给 listeners ：
static void _propagateToListeners(_Future source, _FutureListener listeners) {
while (true) {
assert(source._isComplete);
bool hasError = source._hasError;
if (listeners == null) {
if (hasError) {
AsyncError asyncError = source._error;
source._zone
.handleUncaughtError(asyncError.error, asyncError.stackTrace);
}
return;
}
// Usually futures only have one listener. If they have several, we
// call handle them separately in recursive calls, continuing
// here only when there is only one listener left.
while (listeners._nextListener != null) {
_FutureListener listener = listeners;
listeners = listener._nextListener;
listener._nextListener = null;
_propagateToListeners(source, listener);
}
_FutureListener listener = listeners;
final sourceResult = source._resultOrListeners;
// Do the actual propagation.
// Set initial state of listenerHasError and listenerValueOrError. These
// variables are updated with the outcome of potential callbacks.
// Non-error results, including futures, are stored in
// listenerValueOrError and listenerHasError is set to false. Errors
// are stored in listenerValueOrError as an [AsyncError] and
// listenerHasError is set to true.
bool listenerHasError = hasError;
var listenerValueOrError = sourceResult;
// Only if we either have an error or callbacks, go into this, somewhat
// expensive, branch. Here we'll enter/leave the zone. Many futures
// don't have callbacks, so this is a significant optimization.
if (hasError || listener.handlesValue || listener.handlesComplete) {
Zone zone = listener._zone;
if (hasError && !source._zone.inSameErrorZone(zone)) {
// Don’t cross zone boundaries with errors.
AsyncError asyncError = source._error;
source._zone
.handleUncaughtError(asyncError.error, asyncError.stackTrace);
return;
}
Zone oldZone;
if (!identical(Zone.current, zone)) {
// Change zone if it's not current.
oldZone = Zone._enter(zone);
}
// These callbacks are abstracted to isolate the try/catch blocks
// from the rest of the code to work around a V8 glass jaw.
void handleWhenCompleteCallback() {
// The whenComplete-handler is not combined with normal value/error
// handling. This means at most one handleX method is called per
// listener.
assert(!listener.handlesValue);
assert(!listener.handlesError);
var completeResult;
try {
completeResult = listener.handleWhenComplete();
} catch (e, s) {
if (hasError && identical(source._error.error, e)) {
listenerValueOrError = source._error;
} else {
listenerValueOrError = new AsyncError(e, s);
}
listenerHasError = true;
return;
}
if (completeResult is Future) {
if (completeResult is _Future && completeResult._isComplete) {
if (completeResult._hasError) {
listenerValueOrError = completeResult._error;
listenerHasError = true;
}
// Otherwise use the existing result of source.
return;
}
// We have to wait for the completeResult future to complete
// before knowing if it’s an error or we should use the result
// of source.
var originalSource = source;
listenerValueOrError = completeResult.then((_) => originalSource);
listenerHasError = false;
}
}
void handleValueCallback() {
try {
listenerValueOrError = listener.handleValue(sourceResult);
} catch (e, s) {
listenerValueOrError = new AsyncError(e, s);
listenerHasError = true;
}
}
void handleError() {
try {
AsyncError asyncError = source._error;
if (listener.matchesErrorTest(asyncError) &&
listener.hasErrorCallback) {
listenerValueOrError = listener.handleError(asyncError);
listenerHasError = false;
}
} catch (e, s) {
if (identical(source._error.error, e)) {
listenerValueOrError = source._error;
} else {
listenerValueOrError = new AsyncError(e, s);
}
listenerHasError = true;
}
}
if (listener.handlesComplete) {
handleWhenCompleteCallback();
} else if (!hasError) {
if (listener.handlesValue) {
handleValueCallback();
}
} else {
if (listener.handlesError) {
handleError();
}
}
// If we changed zone, oldZone will not be null.
if (oldZone != null) Zone._leave(oldZone);
// If the listener’s value is a future we need to chain it. Note that
// this can only happen if there is a callback.
if (listenerValueOrError is Future) {
Future chainSource = listenerValueOrError;
// Shortcut if the chain-source is already completed. Just continue
// the loop.
_Future result = listener.result;
if (chainSource is _Future) {
if (chainSource._isComplete) {
listeners = result._removeListeners();
result._cloneResult(chainSource);
source = chainSource;
continue;
} else {
_chainCoreFuture(chainSource, result);
}
} else {
_chainForeignFuture(chainSource, result);
}
return;
}
}
_Future result = listener.result;
listeners = result._removeListeners();
if (!listenerHasError) {
result._setValue(listenerValueOrError);
} else {
AsyncError asyncError = listenerValueOrError;
result._setErrorObject(asyncError);
}
// Prepare for next round.
source = result;
}
}这个函数相对来说比较重量级，其完成的功能也是比较多的，不过结构还是比较清晰的，换算下来就是：
static void _propagateToListeners(_Future source, _FutureListener listeners) {
while (true) {
// 利用递归，将 source 的结果分发给 listeners 中的每一个 listener
// 如果 listener 具有这三个之一的能力，就去根据情况执行其中一个
if (hasError || listener.handlesValue || listener.handlesComplete) {
// 声明 handleWhenCompleteCallback 函数
// 声明 handleValueCallback 函数
// 声明 handleError 函数
// 根据 listener 的能力及 source 的执行结果，选择上面的其中一个函数执行
// 如果 listener 的返回结果还是一个 Future ，那就调用 _chainCoreFuture 或 _chainForeignFuture 进行处理
}
// 如果返回的是一个常规值，则将 value 设置给 result ，接着将 result 的结果分发给 result 的 listeners，所以给 source 和 listeners 重新赋值，while 循环
}
}脉络就是这样。这个函数里面又出现了两种分支，即 listener 处理之后的结果是否是一个 Future ，可以看如下两个示例：
test7() {
print("start");
Future<String> f = Future.delayed(Duration(milliseconds: 100), () {
print("return test7");
return "test7";
});
Future<String> ff = Future(() {
print("return f");
return f;
});
Future(() {
print("return ff");
return ff;
}).then((String s) {
return Future.delayed(Duration(milliseconds: 100), () {
print("return s");
return s;
});
}).then((String s) => print(s));
print("end");
}
test8() {
print("start");
Future.delayed(Duration(milliseconds: 100), () {
print("return test8");
return "test8";
}).then((String s) {
print("return s1");
return s;
}).then((String s) {
print("return s2");
return s;
}).then((String s) {
print("return s3");
return s;
}).then((String s) {
print(s);
});
print("end");
}在 test7 中，Future 的返回值是 ff ，从上面对 _complete 函数的分析得知，test7 可以做如下转换：
test7() {
print("start");
Future<String> f = Future.delayed(Duration(milliseconds: 100), () {
print("return test7");
return "test7";
});
f.then((String s) {
return Future.delayed(Duration(milliseconds: 100), () {
print("return s");
return s;
});
}).then((String s) => print(s));
print("end");
}那么当 f 执行完毕的时候，便会通过 _propagateToListeners 将返回结果传递给 listener ，接着执行 listener 的 callback，但是 listener 的返回结果还是一个 Future ，则会执行这一段代码：
if (listenerValueOrError is Future) {
Future chainSource = listenerValueOrError;
// Shortcut if the chain-source is already completed. Just continue
// the loop.
_Future result = listener.result;
if (chainSource is _Future) {
if (chainSource._isComplete) {
listeners = result._removeListeners();
result._cloneResult(chainSource);
source = chainSource;
continue;
} else {
_chainCoreFuture(chainSource, result);
}
} else {
_chainForeignFuture(chainSource, result);
}
return;
}所以，test7 又会变成如下这样：
test7() {
print("start");
Future<String> f = Future.delayed(Duration(milliseconds: 100), () {
print("return test7");
return "test7";
});
f.then((String s) {
Future f = Future.delayed(Duration(milliseconds: 100), () {
print("return s");
return s;
});
f.then((String s) => print(s));
return f;
});
print("end");
}至此，当 f 中的 Future 也执行完毕的时候，就会出发最后的 listener ，打印 s。 而对于 test8 ，多个 then 连接在一起，且返回的结果都是常规值，则对应着这段代码：
_Future result = listener.result;
listeners = result._removeListeners();
if (!listenerHasError) {
result._setValue(listenerValueOrError);
} else {
AsyncError asyncError = listenerValueOrError;
result._setErrorObject(asyncError);
}
// Prepare for next round.
source = result;result 即是 listener 对应的 Future，listeners 是 result 的 listeners，所以，每当一个 while 循环执行之后，就意味着一个 then 函数的结束，直到所有的 listener 都得到处理。 再来看下 then 函数：
Future<R> then<R>(FutureOr<R> f(T value), {Function onError}) {
Zone currentZone = Zone.current;
if (!identical(currentZone, _rootZone)) {
f = currentZone.registerUnaryCallback<FutureOr<R>, T>(f);
if (onError != null) {
// In checked mode, this checks that onError is assignable to one of:
// dynamic Function(Object)
// dynamic Function(Object, StackTrace)
onError = _registerErrorHandler(onError, currentZone);
}
}
_Future<R> result = new _Future<R>();
_addListener(new _FutureListener<T, R>.then(result, f, onError));
return result;
}_addListener 函数可以看出，Future 的 listener 就是在这个函数中添加的，listener 对应的 Future ，也是在这里创建的，result、callback、onError 将会被构造成 _FutureListener 实例，_addListener 负责将 listener 添加到 Future 上。
void _addListener(_FutureListener listener) {
assert(listener._nextListener == null);
if (_mayAddListener) {
listener._nextListener = _resultOrListeners;
_resultOrListeners = listener;
} else {
if (_isChained) {
// Delegate listeners to chained source future.
// If the source is complete, instead copy its values and
// drop the chaining.
_Future source = _chainSource;
if (!source._isComplete) {
source._addListener(listener);
return;
}
_cloneResult(source);
}
assert(_isComplete);
// Handle late listeners asynchronously.
_zone.scheduleMicrotask(() {
_propagateToListeners(this, listener);
});
}
}一般情况下，listener 会被添加到 Future 的 listeners 中，当 Future 已经是 Future 链中的一员时，listener 则会直接被添加到 source 中去，或者当 Future 已经执行完成时，则直接调用 _propagateToListeners 处理。 再来说说 Future 中常用的一些参数，比如 _isChained 、_isComplete 以及 _mayAddListener 这些到底是怎么确定的？_chainSource 、_resultOrListeners 这些变量都是些什么？ 总的来说，这些都指向两个变量，_state 和 _resultOrListeners，比如：
bool get _mayComplete => _state == _stateIncomplete;
bool get _isPendingComplete => _state == _statePendingComplete;
bool get _mayAddListener => _state <= _statePendingComplete;
bool get _isChained => _state == _stateChained;
bool get _isComplete => _state >= _stateValue;
bool get _hasError => _state == _stateError;又比如：
AsyncError get _error {
assert(_hasError);
return _resultOrListeners;
}
_Future get _chainSource {
assert(_isChained);
return _resultOrListeners;
}
_FutureListener _removeListeners() {
// Reverse listeners before returning them, so the resulting list is in
// subscription order.
assert(!_isComplete);
_FutureListener current = _resultOrListeners;
_resultOrListeners = null;
return _reverseListeners(current);
}
void _setChained(_Future source) {
assert(_mayAddListener);
_state = _stateChained;
_resultOrListeners = source;
}
void _setValue(T value) {
assert(!_isComplete); // But may have a completion pending.
_state = _stateValue;
_resultOrListeners = value;
}
void _setErrorObject(AsyncError error) {
assert(!_isComplete); // But may have a completion pending.
_state = _stateError;
_resultOrListeners = error;
}
void _cloneResult(_Future source) {
assert(!_isComplete);
assert(source._isComplete);
_state = source._state;
_resultOrListeners = source._resultOrListeners;
}所以，error 、result、chainSource、listeners 实际上都是同一个变量表示的，只不过在不同的状态下，_resultOrListeners 有着不同的含义，在 _hasError 状态下，_resultOrListeners 表示 error，在 !_isComplete 状态下，_resultOrListeners 表示 listeners，而在给 _resultOrListeners 设置值的时候，一般也会一并给 _state 赋值。