Flutter 异步机制：Isolate
在 Future 一篇中有说到，Future 本质上是单线程异步，一般只适合非耗时任务的执行，否则就会堵塞后续的 UI 操作，而耗时任务，还是需要使用多线程才行，在 flutter 中没有 Thread ，但是可以通过 Isolate 开启新的线程，并且对多线程进行了封装，使得线程之间的数据不能互通，而需要使用 ReceivePort 和 SendPort 进行通信。比如：
test1() {
ReceivePort receivePort = ReceivePort();
ReceivePort exitPort = ReceivePort();
ReceivePort errorPort = ReceivePort();
Isolate.spawn(test1EntryPoint, receivePort.sendPort, debugName: "debug test9",
onExit: exitPort.sendPort, onError: errorPort.sendPort).then((_) {
print("spawn end");
});
receivePort.listen((message) {
print(message);
}, onDone: () {
print("on done");
});
exitPort.listen((_) {
print("exit");
receivePort.close();
exitPort.close();
errorPort.close();
});
errorPort.listen((_) {
print("error");
});
}
test1EntryPoint(SendPort sendPort) {
// Timer.periodic(Duration(seconds: 1), (Timer timer) {
// sendPort.send("test1");
// });
Timer(Duration(seconds: 1), () {
sendPort.send("test1");
});
}开启线程需要使用 Isolate.spawn 函数，这样一个函数大致需要两个过程：
- 创建一个 Isolate 实例
- 在这个 Isolate 中通过 ReceivePort 与原线程通信
开启 Isolate
// third_party/sdk/runtime/lib/isolate_patch.dart
@patch
static Future<Isolate> spawn<T>(void entryPoint(T message), T message,
{bool paused: false,
bool errorsAreFatal,
SendPort onExit,
SendPort onError,
String debugName}) async {
// `paused` isn't handled yet.
RawReceivePort readyPort;
try {
// Check for the type of `entryPoint` on the spawning isolate to make
// error-handling easier.
if (entryPoint is! _UnaryFunction) {
throw new ArgumentError(entryPoint);
}
// The VM will invoke [_startIsolate] with entryPoint as argument.
readyPort = new RawReceivePort();
// We do not inherit the package config settings from the parent isolate,
// instead we use the values that were set on the command line.
var packageConfig = VMLibraryHooks.packageConfigString;
var script = VMLibraryHooks.platformScript;
if (script == null) {
// We do not have enough information to support spawning the new
// isolate.
throw new UnsupportedError("Isolate.spawn");
}
if (script.scheme == "package") {
script = await Isolate.resolvePackageUri(script);
}
_spawnFunction(
readyPort.sendPort,
script.toString(),
entryPoint,
message,
paused,
errorsAreFatal,
onExit,
onError,
null,
packageConfig,
debugName);
return await _spawnCommon(readyPort);
} catch (e, st) {
if (readyPort != null) {
readyPort.close();
}
return await new Future<Isolate>.error(e, st);
}
}检查传入的 entryPoint 函数是否符合要求（只能有一个参数），然后取出创建 Isolate 所需的配置，再调用 _spawnFunction 继续执行，最后以 _spawnCommon 收尾。
_spawnFunction 是一个 native 函数，对应的 c++ 的实现为：
DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 0, 11) {
GET_NON_NULL_NATIVE_ARGUMENT(SendPort, port, arguments->NativeArgAt(0));
GET_NON_NULL_NATIVE_ARGUMENT(String, script_uri, arguments->NativeArgAt(1));
GET_NON_NULL_NATIVE_ARGUMENT(Instance, closure, arguments->NativeArgAt(2));
GET_NON_NULL_NATIVE_ARGUMENT(Instance, message, arguments->NativeArgAt(3));
GET_NON_NULL_NATIVE_ARGUMENT(Bool, paused, arguments->NativeArgAt(4));
GET_NATIVE_ARGUMENT(Bool, fatalErrors, arguments->NativeArgAt(5));
GET_NATIVE_ARGUMENT(SendPort, onExit, arguments->NativeArgAt(6));
GET_NATIVE_ARGUMENT(SendPort, onError, arguments->NativeArgAt(7));
GET_NATIVE_ARGUMENT(String, packageRoot, arguments->NativeArgAt(8));
GET_NATIVE_ARGUMENT(String, packageConfig, arguments->NativeArgAt(9));
GET_NATIVE_ARGUMENT(String, debugName, arguments->NativeArgAt(10));
if (closure.IsClosure()) {
Function& func = Function::Handle();
func = Closure::Cast(closure).function();
if (func.IsImplicitClosureFunction() && func.is_static()) {
#if defined(DEBUG)
Context& ctx = Context::Handle();
ctx = Closure::Cast(closure).context();
ASSERT(ctx.IsNull());
#endif
// Get the parent function so that we get the right function name.
func = func.parent_function();
bool fatal_errors = fatalErrors.IsNull() ? true : fatalErrors.value();
Dart_Port on_exit_port = onExit.IsNull() ? ILLEGAL_PORT : onExit.Id();
Dart_Port on_error_port = onError.IsNull() ? ILLEGAL_PORT : onError.Id();
// We first try to serialize the message. In case the message is not
// serializable this will throw an exception.
SerializedObjectBuffer message_buffer;
{
MessageWriter writer(/* can_send_any_object = */ true);
message_buffer.set_message(writer.WriteMessage(
message, ILLEGAL_PORT, Message::kNormalPriority));
}
const char* utf8_package_config =
packageConfig.IsNull() ? NULL : String2UTF8(packageConfig);
const char* utf8_debug_name =
debugName.IsNull() ? NULL : String2UTF8(debugName);
std::unique_ptr<IsolateSpawnState> state(new IsolateSpawnState(
port.Id(), isolate->origin_id(), String2UTF8(script_uri), func,
&message_buffer, utf8_package_config, paused.value(), fatal_errors,
on_exit_port, on_error_port, utf8_debug_name, isolate->group()));
// Since this is a call to Isolate.spawn, copy the parent isolate's code.
state->isolate_flags()->copy_parent_code = true;
Dart::thread_pool()->Run<SpawnIsolateTask>(isolate, std::move(state));
return Object::null();
}
}
const String& msg = String::Handle(String::New(
"Isolate.spawn expects to be passed a static or top-level function"));
Exceptions::ThrowArgumentError(msg);
return Object::null();
}首先还是对 entryPoint 函数的检测，要求其是一个闭包（静态函数，或非内部函数），然后依次将 entryPoint 转换成 func，onExit 和 onError 都找到其对应的 Dart_Port ，将 message 序列化，创建出 IsolateSpawnState 实例和 SpawnIsolateTask 实例，最后调用线程池的 Run 启动 Isolate 。
// third_party/sdk/runtime/vm/isolate.cc
IsolateSpawnState::IsolateSpawnState(Dart_Port parent_port,
Dart_Port origin_id,
const char* script_url,
const Function& func,
SerializedObjectBuffer* message_buffer,
const char* package_config,
bool paused,
bool errors_are_fatal,
Dart_Port on_exit_port,
Dart_Port on_error_port,
const char* debug_name,
IsolateGroup* isolate_group)
: isolate_(nullptr),
parent_port_(parent_port),
origin_id_(origin_id),
on_exit_port_(on_exit_port),
on_error_port_(on_error_port),
script_url_(script_url),
package_config_(package_config),
library_url_(nullptr),
class_name_(nullptr),
function_name_(nullptr),
debug_name_(debug_name),
isolate_group_(isolate_group),
serialized_args_(nullptr),
serialized_message_(message_buffer->StealMessage()),
paused_(paused),
errors_are_fatal_(errors_are_fatal) {
const Class& cls = Class::Handle(func.Owner());
const Library& lib = Library::Handle(cls.library());
const String& lib_url = String::Handle(lib.url());
library_url_ = NewConstChar(lib_url.ToCString());
String& func_name = String::Handle();
func_name = func.name();
func_name = String::ScrubName(func_name);
function_name_ = NewConstChar(func_name.ToCString());
if (!cls.IsTopLevel()) {
const String& class_name = String::Handle(cls.Name());
class_name_ = NewConstChar(class_name.ToCString());
}
// Inherit flags from spawning isolate.
Isolate::Current()->FlagsCopyTo(isolate_flags());
}由此可见，IsolateSpawnState 就是一个存储工具。SpawnIsolateTask 继承自 ThreadPool::Task，实现了 Run 函数，这个函数会在后面被调用到。
ThreadPool:Run
// third_party/sdk/runtime/vm/thread_pool.h
bool Run(Args&&... args) {
return RunImpl(std::unique_ptr<Task>(new T(std::forward<Args>(args)...)));
}
// third_party/sdk/runtime/vm/thread_pool.cc
bool ThreadPool::RunImpl(std::unique_ptr<Task> task) {
Worker* worker = NULL;
bool new_worker = false;
{
// We need ThreadPool::mutex_ to access worker lists and other
// ThreadPool state.
MutexLocker ml(&mutex_);
if (shutting_down_) {
return false;
}
if (idle_workers_ == NULL) {
worker = new Worker(this);
ASSERT(worker != NULL);
new_worker = true;
count_started_++;
// Add worker to the all_workers_ list.
worker->all_next_ = all_workers_;
all_workers_ = worker;
worker->owned_ = true;
count_running_++;
} else {
// Get the first worker from the idle worker list.
worker = idle_workers_;
idle_workers_ = worker->idle_next_;
worker->idle_next_ = NULL;
count_idle_--;
count_running_++;
}
}
// Release ThreadPool::mutex_ before calling Worker functions.
ASSERT(worker != NULL);
worker->SetTask(std::move(task));
if (new_worker) {
// Call StartThread after we've assigned the first task.
worker->StartThread();
}
return true;
}Worker 可以表示线程池中的一个线程，在这个函数中，如果没有空闲的 worker ，则创建一个新的 worker ，然后将 task 传给 worker ，在 worker 中调度执行 task ，如果是一个新创建的 worker ，还需要调用 StartThread 创建线程并启动执行。
// third_party/sdk/runtime/vm/thread_pool.cc
void ThreadPool::Worker::StartThread() {
#if defined(DEBUG)
// Must call SetTask before StartThread.
{ // NOLINT
MonitorLocker ml(&monitor_);
ASSERT(task_ != nullptr);
}
#endif
int result = OSThread::Start("Dart ThreadPool Worker", &Worker::Main,
reinterpret_cast<uword>(this));
if (result != 0) {
FATAL1("Could not start worker thread: result = %d.", result);
}
}OSThread::Start 函数会开启一个线程，并在该线程中执行 Worker::Main 函数，并将 work 作为参数传递进去。
// third_party/sdk/runtime/vm/thread_pool.cc
void ThreadPool::Worker::Main(uword args) {
Worker* worker = reinterpret_cast<Worker*>(args);
OSThread* os_thread = OSThread::Current();
ASSERT(os_thread != NULL);
ThreadId id = os_thread->id();
ThreadPool* pool;
// Set the thread's stack_base based on the current stack pointer.
os_thread->RefineStackBoundsFromSP(OSThread::GetCurrentStackPointer());
{
MonitorLocker ml(&worker->monitor_);
ASSERT(worker->task_);
worker->id_ = id;
pool = worker->pool_;
}
bool released = worker->Loop();
// It should be okay to access these unlocked here in this assert.
// worker->all_next_ is retained by the pool for shutdown monitoring.
ASSERT(!worker->owned_ && (worker->idle_next_ == NULL));
if (!released) {
// This worker is exiting because the thread pool is being shut down.
// Inform the thread pool that we are exiting. We remove this worker from
// shutting_down_workers_ list because there will be no need for the
// ThreadPool to take action for this worker.
ThreadJoinId join_id = OSThread::GetCurrentThreadJoinId(os_thread);
{
MutexLocker ml(&pool->mutex_);
JoinList::AddLocked(join_id, &pool->join_list_);
}
// worker->id_ should never be read again, so set to invalid in debug mode
// for asserts.
#if defined(DEBUG)
{
MonitorLocker ml(&worker->monitor_);
worker->id_ = OSThread::kInvalidThreadId;
}
#endif
// Remove from the shutdown list, delete, and notify the thread pool.
{
MonitorLocker eml(&pool->exit_monitor_);
pool->RemoveWorkerFromShutdownList(worker);
delete worker;
eml.Notify();
}
} else {
// This worker is going down because it was idle for too long. This case
// is not due to a ThreadPool Shutdown. Thus, we simply delete the worker.
// The worker's id is added to the thread pool's join list by
// ReleaseIdleWorker, so in the case that the thread pool begins shutting
// down immediately after returning from worker->Loop() above, we still
// wait for the thread to exit by joining on it in Shutdown().
delete worker;
}
// Call the thread exit hook here to notify the embedder that the
// thread pool thread is exiting.
if (Dart::thread_exit_callback() != NULL) {
(*Dart::thread_exit_callback())();
}
}Loop 函数用于执行 task ，当 Loop 退出的时候，worker 的工作就结束了，会被释放、删除。
bool ThreadPool::Worker::Loop() {
MonitorLocker ml(&monitor_);
int64_t idle_start;
while (true) {
ASSERT(task_ != nullptr);
std::unique_ptr<Task> task = std::move(task_);
// Release monitor while handling the task.
ml.Exit();
task->Run();
ASSERT(Isolate::Current() == NULL);
task.reset();
ml.Enter();
ASSERT(task_ == nullptr);
if (IsDone()) {
return false;
}
ASSERT(!done_);
pool_->SetIdleAndReapExited(this);
idle_start = OS::GetCurrentMonotonicMicros();
while (true) {
Monitor::WaitResult result = ml.WaitMicros(ComputeTimeout(idle_start));
if (task_ != nullptr) {
// We've found a task. Process it, regardless of whether the
// worker is done_.
break;
}
if (IsDone()) {
return false;
}
if ((result == Monitor::kTimedOut) && pool_->ReleaseIdleWorker(this)) {
return true;
}
}
}
UNREACHABLE();
return false;
}首先拿出 task 并执行，执行完 worker 进入空闲状态，SetIdleAndReapExited 将 worker 加入空闲队列，等待一段时间之后如果还没有新的 task 到来，则推出循环，释放 worker 。而 task 的执行就是调用其 Run 函数，也就是上面说到的 SpawnIsolateTask 的 Run 。
SpawnIsolateTask:Run
// third_party/sdk/runtime/lib/isolate.cc
void Run() override {
auto group = state_->isolate_group();
// The create isolate group call back is mandatory. If not provided we
// cannot spawn isolates.
Dart_IsolateGroupCreateCallback create_group_callback =
Isolate::CreateGroupCallback();
if (create_group_callback == nullptr) {
FailedSpawn("Isolate spawn is not supported by this Dart embedder\n");
return;
}
// The initialize callback is optional atm, we fall back to creating isolate
// groups if it was not provided.
Dart_InitializeIsolateCallback initialize_callback =
Isolate::InitializeCallback();
const char* name = (state_->debug_name() == NULL) ? state_->function_name()
: state_->debug_name();
ASSERT(name != NULL);
// Create a new isolate.
char* error = nullptr;
Isolate* isolate = nullptr;
if (!FLAG_enable_isolate_groups || group == nullptr ||
initialize_callback == nullptr) {
// Make a copy of the state's isolate flags and hand it to the callback.
Dart_IsolateFlags api_flags = *(state_->isolate_flags());
isolate = reinterpret_cast<Isolate*>((create_group_callback)(
state_->script_url(), name, nullptr, state_->package_config(),
&api_flags, parent_isolate_->init_callback_data(), &error));
parent_isolate_->DecrementSpawnCount();
parent_isolate_ = nullptr;
} else {
if (initialize_callback == nullptr) {
FailedSpawn("Isolate spawn is not supported by this embedder.");
return;
}
isolate = CreateWithinExistingIsolateGroup(group, name, &error);
parent_isolate_->DecrementSpawnCount();
parent_isolate_ = nullptr;
if (isolate == nullptr) {
FailedSpawn(error);
free(error);
return;
}
void* child_isolate_data = nullptr;
bool success = initialize_callback(&child_isolate_data, &error);
isolate->set_init_callback_data(child_isolate_data);
if (!success) {
Dart_ShutdownIsolate();
FailedSpawn(error);
free(error);
return;
}
Dart_ExitIsolate();
}
if (isolate == nullptr) {
FailedSpawn(error);
free(error);
return;
}
if (state_->origin_id() != ILLEGAL_PORT) {
// For isolates spawned using spawnFunction we set the origin_id
// to the origin_id of the parent isolate.
isolate->set_origin_id(state_->origin_id());
}
MutexLocker ml(isolate->mutex());
state_->set_isolate(isolate);
isolate->set_spawn_state(std::move(state_));
if (isolate->is_runnable()) {
isolate->Run();
}
}此时的线程已经创建完成，但是对应的 Isolate 还没有创建，该函数一开始就获取 create_group_callback 函数，后面会看到就是调用这个函数生成的 Isolate 实例，最后调用 Isolate 的 Run 函数。共两个部分：
- 初始化 Isolate
- 执行 Isolate
初始化 Isolate
初始化 Isolate 有两种方式，如果满足以下条件：
if (!FLAG_enable_isolate_groups || group == nullptr || initialize_callback == nullptr)则调用 create_group_callback 创建，否则就调用 CreateWithinExistingIsolateGroup 创建 Isolate ，并使用 initialize_callback 对其初始化。create_group_callback 是在 DartVM 初始化的时候设置的，然后调用 Dart_Initialize、Dart:Init 等一步步设置到 Isolate 中，其对应的函数为 DartIsolateGroupCreateCallback：
// runtime/dart_isolate.cc
Dart_Isolate DartIsolate::DartIsolateGroupCreateCallback(
const char* advisory_script_uri,
const char* advisory_script_entrypoint,
const char* package_root,
const char* package_config,
Dart_IsolateFlags* flags,
std::shared_ptr<DartIsolate>* parent_embedder_isolate,
char** error) {
if (parent_embedder_isolate == nullptr &&
strcmp(advisory_script_uri, DART_VM_SERVICE_ISOLATE_NAME) == 0) {
// The VM attempts to start the VM service for us on |Dart_Initialize|. In
// such a case, the callback data will be null and the script URI will be
// DART_VM_SERVICE_ISOLATE_NAME. In such cases, we just create the service
// isolate like normal but dont hold a reference to it at all. We also start
// this isolate since we will never again reference it from the engine.
return DartCreateAndStartServiceIsolate(package_root, //
package_config, //
flags, //
error //
);
}
return CreateDartVMAndEmbedderObjectPair(
advisory_script_uri, // URI
advisory_script_entrypoint, // entrypoint
package_root, // package root
package_config, // package config
flags, // isolate flags
parent_embedder_isolate, // embedder data
false, // is root isolate
error // error
)
.first;
}
std::pair<Dart_Isolate, std::weak_ptr<DartIsolate>>
DartIsolate::CreateDartVMAndEmbedderObjectPair(
const char* advisory_script_uri,
const char* advisory_script_entrypoint,
const char* package_root,
const char* package_config,
Dart_IsolateFlags* flags,
std::shared_ptr<DartIsolate>* p_parent_embedder_isolate,
bool is_root_isolate,
char** error) {
TRACE_EVENT0("flutter", "DartIsolate::CreateDartVMAndEmbedderObjectPair");
std::unique_ptr<std::shared_ptr<DartIsolate>> embedder_isolate(
p_parent_embedder_isolate);
if (embedder_isolate == nullptr) {
*error =
strdup("Parent isolate did not have embedder specific callback data.");
FML_DLOG(ERROR) << *error;
return {nullptr, {}};
}
if (!is_root_isolate) {
auto* raw_embedder_isolate = embedder_isolate.release();
TaskRunners null_task_runners(advisory_script_uri, nullptr, nullptr,
nullptr, nullptr);
// Copy most fields from the parent to the child.
embedder_isolate = std::make_unique<std::shared_ptr<DartIsolate>>(
std::shared_ptr<DartIsolate>(new DartIsolate(
(*raw_embedder_isolate)->GetSettings(), // settings
(*raw_embedder_isolate)->GetIsolateSnapshot(), // isolate_snapshot
null_task_runners, // task_runners
fml::WeakPtr<IOManager>{}, // io_manager
fml::RefPtr<SkiaUnrefQueue>{}, // unref_queue
fml::WeakPtr<ImageDecoder>{}, // image_decoder
advisory_script_uri, // advisory_script_uri
advisory_script_entrypoint, // advisory_script_entrypoint
(*raw_embedder_isolate)->child_isolate_preparer_, // preparer
(*raw_embedder_isolate)->isolate_create_callback_, // on create
(*raw_embedder_isolate)->isolate_shutdown_callback_ // on shutdown
))
);
}
// Create the Dart VM isolate and give it the embedder object as the baton.
Dart_Isolate isolate = Dart_CreateIsolateGroup(
advisory_script_uri, //
advisory_script_entrypoint, //
(*embedder_isolate)->GetIsolateSnapshot()->GetDataMapping(),
(*embedder_isolate)->GetIsolateSnapshot()->GetInstructionsMapping(),
flags,
embedder_isolate.get(), // isolate_group_data
embedder_isolate.get(), // isolate_group
error);
if (isolate == nullptr) {
FML_DLOG(ERROR) << *error;
return {nullptr, {}};
}
if (!(*embedder_isolate)->Initialize(isolate, is_root_isolate)) {
*error = strdup("Embedder could not initialize the Dart isolate.");
FML_DLOG(ERROR) << *error;
return {nullptr, {}};
}
if (!(*embedder_isolate)->LoadLibraries(is_root_isolate)) {
*error =
strdup("Embedder could not load libraries in the new Dart isolate.");
FML_DLOG(ERROR) << *error;
return {nullptr, {}};
}
auto weak_embedder_isolate = (*embedder_isolate)->GetWeakIsolatePtr();
// Root isolates will be setup by the engine and the service isolate (which is
// also a root isolate) by the utility routines in the VM. However, secondary
// isolates will be run by the VM if they are marked as runnable.
if (!is_root_isolate) {
FML_DCHECK((*embedder_isolate)->child_isolate_preparer_);
if (!(*embedder_isolate)
->child_isolate_preparer_((*embedder_isolate).get())) {
*error = strdup("Could not prepare the child isolate to run.");
FML_DLOG(ERROR) << *error;
return {nullptr, {}};
}
}
// The ownership of the embedder object is controlled by the Dart VM. So the
// only reference returned to the caller is weak.
embedder_isolate.release();
return {isolate, weak_embedder_isolate};
}依次创建 DartIsolate 和 Dart_Isolate，在 DartIsolate 的构造函数中还创建了 UIDartState 实例，用于保存一些数据，然后调用 Dart_CreateIsolateGroup 进行 Isolate 创建过程。
// third_party/sdk/runtime/vm/dart_api_impl.cc
DART_EXPORT Dart_Isolate
Dart_CreateIsolateGroup(const char* script_uri,
const char* name,
const uint8_t* snapshot_data,
const uint8_t* snapshot_instructions,
const uint8_t* shared_data,
const uint8_t* shared_instructions,
Dart_IsolateFlags* flags,
void* isolate_group_data,
void* isolate_data,
char** error) {
API_TIMELINE_DURATION(Thread::Current());
Dart_IsolateFlags api_flags;
if (flags == nullptr) {
Isolate::FlagsInitialize(&api_flags);
flags = &api_flags;
}
const char* non_null_name = name == nullptr ? "isolate" : name;
std::unique_ptr<IsolateGroupSource> source(new IsolateGroupSource(
script_uri, non_null_name, snapshot_data, snapshot_instructions,
shared_data, shared_instructions, nullptr, -1, *flags));
auto group = new IsolateGroup(std::move(source), isolate_group_data);
Dart_Isolate isolate =
CreateIsolate(group, non_null_name, isolate_data, error);
if (isolate != nullptr) {
group->set_initial_spawn_successful();
}
return isolate;
}三个小步骤，分别是创建 IsolateGroupSource ，创建 IsolateGroup，最后创建 Isolate ，IsolateGroupSource 是 IsolateGroup 持有的一些基本信息，IsolateGroup 将 Isolate 分组，IsolateGroup 对其进行一些管理，同时同组的 Isolate 也可以共享 IsolateGroup 的数据。
// third_party/sdk/runtime/vm/dart_api_impl.cc
static Dart_Isolate CreateIsolate(IsolateGroup* group,
const char* name,
void* isolate_data,
char** error) {
CHECK_NO_ISOLATE(Isolate::Current());
auto source = group->source();
Isolate* I = Dart::CreateIsolate(name, source->flags, group);
if (I == NULL) {
if (error != NULL) {
*error = strdup("Isolate creation failed");
}
return reinterpret_cast<Dart_Isolate>(NULL);
}
Thread* T = Thread::Current();
bool success = false;
{
StackZone zone(T);
HANDLESCOPE(T);
// We enter an API scope here as InitializeIsolate could compile some
// bootstrap library files which call out to a tag handler that may create
// Api Handles when an error is encountered.
T->EnterApiScope();
const Error& error_obj =
Error::Handle(Z, Dart::InitializeIsolate(
source->snapshot_data,
source->snapshot_instructions, source->shared_data,
source->shared_instructions, source->kernel_buffer,
source->kernel_buffer_size, isolate_data));
if (error_obj.IsNull()) {
#if defined(DART_NO_SNAPSHOT) && !defined(PRODUCT)
if (FLAG_check_function_fingerprints && source->kernel_buffer == NULL) {
Library::CheckFunctionFingerprints();
}
#endif // defined(DART_NO_SNAPSHOT) && !defined(PRODUCT).
success = true;
} else if (error != NULL) {
*error = strdup(error_obj.ToErrorCString());
}
// We exit the API scope entered above.
T->ExitApiScope();
}
if (success) {
// A Thread structure has been associated to the thread, we do the
// safepoint transition explicitly here instead of using the
// TransitionXXX scope objects as the reverse transition happens
// outside this scope in Dart_ShutdownIsolate/Dart_ExitIsolate.
T->set_execution_state(Thread::kThreadInNative);
T->EnterSafepoint();
if (error != NULL) {
*error = NULL;
}
return Api::CastIsolate(I);
}
Dart::ShutdownIsolate();
return reinterpret_cast<Dart_Isolate>(NULL);
}该函数先调用 CreateIsolate 创建出 Isolate 实例，然后调用 InitializeIsolate 进行初始化。
// third_party/sdk/runtime/vm/dart.cc
Isolate* Dart::CreateIsolate(const char* name_prefix,
const Dart_IsolateFlags& api_flags,
IsolateGroup* isolate_group) {
// Create a new isolate.
Isolate* isolate =
Isolate::InitIsolate(name_prefix, isolate_group, api_flags);
return isolate;
}
// 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) {
Isolate* result = new Isolate(isolate_group, api_flags);
ASSERT(result != nullptr);
#if !defined(PRODUCT)
// Initialize metrics.
#define ISOLATE_METRIC_INIT(type, variable, name, unit) \
result->metric_##variable##_.InitInstance(result, name, NULL, Metric::unit);
ISOLATE_METRIC_LIST(ISOLATE_METRIC_INIT);
#undef ISOLATE_METRIC_INIT
#endif // !defined(PRODUCT)
bool is_service_or_kernel_isolate = false;
if (ServiceIsolate::NameEquals(name_prefix)) {
ASSERT(!ServiceIsolate::Exists());
is_service_or_kernel_isolate = true;
}
#if !defined(DART_PRECOMPILED_RUNTIME)
if (KernelIsolate::NameEquals(name_prefix)) {
ASSERT(!KernelIsolate::Exists());
KernelIsolate::SetKernelIsolate(result);
is_service_or_kernel_isolate = true;
}
#endif // !defined(DART_PRECOMPILED_RUNTIME)
Heap::Init(result,
is_vm_isolate
? 0 // New gen size 0; VM isolate should only allocate in old.
: FLAG_new_gen_semi_max_size * MBInWords,
(is_service_or_kernel_isolate ? kDefaultMaxOldGenHeapSize
: FLAG_old_gen_heap_size) *
MBInWords);
// TODO(5411455): For now just set the recently created isolate as
// the current isolate.
if (!Thread::EnterIsolate(result)) {
// We failed to enter the isolate, it is possible the VM is shutting down,
// return back a NULL so that CreateIsolate reports back an error.
if (KernelIsolate::IsKernelIsolate(result)) {
KernelIsolate::SetKernelIsolate(nullptr);
}
if (ServiceIsolate::IsServiceIsolate(result)) {
ServiceIsolate::SetServiceIsolate(nullptr);
}
delete result;
return nullptr;
}
// Setup the isolate message handler.
MessageHandler* handler = new IsolateMessageHandler(result);
ASSERT(handler != nullptr);
result->set_message_handler(handler);
// Setup the Dart API state.
ApiState* state = new ApiState();
ASSERT(state != nullptr);
result->set_api_state(state);
result->set_main_port(PortMap::CreatePort(result->message_handler()));
#if defined(DEBUG)
// Verify that we are never reusing a live origin id.
VerifyOriginId id_verifier(result->main_port());
Isolate::VisitIsolates(&id_verifier);
#endif
result->set_origin_id(result->main_port());
result->set_pause_capability(result->random()->NextUInt64());
result->set_terminate_capability(result->random()->NextUInt64());
result->BuildName(name_prefix);
#if !defined(PRODUCT)
result->debugger_ = new Debugger(result);
#endif
if (FLAG_trace_isolates) {
if (name_prefix == nullptr || strcmp(name_prefix, "vm-isolate") != 0) {
OS::PrintErr(
"[+] Starting isolate:\n"
"\tisolate: %s\n",
result->name());
}
}
#ifndef PRODUCT
if (FLAG_support_service) {
ObjectIdRing::Init(result);
}
#endif // !PRODUCT
// Add to isolate list. Shutdown and delete the isolate on failure.
if (!AddIsolateToList(result)) {
result->LowLevelShutdown();
Thread::ExitIsolate();
if (KernelIsolate::IsKernelIsolate(result)) {
KernelIsolate::SetKernelIsolate(nullptr);
}
if (ServiceIsolate::IsServiceIsolate(result)) {
ServiceIsolate::SetServiceIsolate(nullptr);
}
delete result;
return nullptr;
}
return result;
}Dart::CreateIsolate 又转而去调用 Isolate::InitIsolate 函数，这里首先是调用 Isolate 的构造函数，初始化了一些参数，并将 Isolate 加入到 IsolateGroup 中，然后就进行了 Heap 的初始化，MessageHandler 初始化等，Thread::EnterIsolate 使线程与 Isolate 建立联系，最后调用 AddIsolateToList 将 Isolate 加入到 Isolate 表中。
// third_party/sdk/runtime/vm/thread.cc
bool Thread::EnterIsolate(Isolate* isolate) {
const bool kIsMutatorThread = true;
Thread* thread = isolate->ScheduleThread(kIsMutatorThread);
if (thread != NULL) {
ASSERT(thread->store_buffer_block_ == NULL);
thread->task_kind_ = kMutatorTask;
thread->StoreBufferAcquire();
if (isolate->marking_stack() != NULL) {
// Concurrent mark in progress. Enable barrier for this thread.
thread->MarkingStackAcquire();
thread->DeferredMarkingStackAcquire();
}
return true;
}
return false;
}调用 ScheduleThread 创建出了 Thread 对象，与之相关的还有 OS_Thread 和 Isolate ，它们都从某种意义上表示一个线程，其中 OS_Thread 可以说是操作系统上的线程，也就是真正意义上的线程，Thread 是 DartVM 中的线程的表示，是对 OS_Thread 的封装，直接服务于 Isolate ，是 Isolate 与线程之间的粘合剂。
// third_party/sdk/runtime/vm/isolate.cc
Thread* Isolate::ScheduleThread(bool is_mutator, bool bypass_safepoint) {
// We are about to associate the thread with an isolate group and it would
// not be possible to correctly track no_safepoint_scope_depth for the
// thread in the constructor/destructor of MonitorLocker,
// so we create a MonitorLocker object which does not do any
// no_safepoint_scope_depth increments/decrements.
MonitorLocker ml(group()->threads_lock(), false);
// Check to make sure we don't already have a mutator thread.
if (is_mutator && scheduled_mutator_thread_ != nullptr) {
return nullptr;
}
// NOTE: We cannot just use `Dart::vm_isolate() == this` here, since during
// VM startup it might not have been set at this point.
const bool is_vm_isolate =
Dart::vm_isolate() == nullptr || Dart::vm_isolate() == this;
// We lazily create a [Thread] structure for the mutator thread, but we'll
// reuse it until the death of the isolate.
Thread* existing_mutator_thread = is_mutator ? mutator_thread_ : nullptr;
// Schedule the thread into the isolate by associating a 'Thread' structure
// with it (this is done while we are holding the thread registry lock).
Thread* thread =
group()->ScheduleThreadLocked(&ml, existing_mutator_thread, is_vm_isolate,
is_mutator, bypass_safepoint);
if (is_mutator) {
ASSERT(mutator_thread_ == nullptr || mutator_thread_ == thread);
mutator_thread_ = thread;
scheduled_mutator_thread_ = thread;
}
thread->isolate_ = this;
ASSERT(heap() != nullptr);
thread->heap_ = heap();
return thread;
}调用 ScheduleThreadLocked 创建 Thread ，然后将 Isolate 和 Heap 保存到 Thread 中。
// third_party/sdk/runtime/vm/isolate.cc
Thread* IsolateGroup::ScheduleThreadLocked(MonitorLocker* ml,
Thread* existing_mutator_thread,
bool is_vm_isolate,
bool is_mutator,
bool bypass_safepoint) {
ASSERT(threads_lock()->IsOwnedByCurrentThread());
// Schedule the thread into the isolate group by associating
// a 'Thread' structure with it (this is done while we are holding
// the thread registry lock).
Thread* thread = nullptr;
OSThread* os_thread = OSThread::Current();
if (os_thread != nullptr) {
// If a safepoint operation is in progress wait for it
// to finish before scheduling this thread in.
while (!bypass_safepoint && safepoint_handler()->SafepointInProgress()) {
ml->Wait();
}
if (is_mutator) {
if (existing_mutator_thread == nullptr) {
// Allocate a new [Thread] structure for the mutator thread.
thread = thread_registry()->GetFreeThreadLocked(is_vm_isolate);
} else {
// Reuse the existing cached [Thread] structure for the mutator thread.,
// see comment in 'base_isolate.h'.
thread_registry()->AddToActiveListLocked(existing_mutator_thread);
thread = existing_mutator_thread;
}
} else {
thread = thread_registry()->GetFreeThreadLocked(is_vm_isolate);
}
// Now get a free Thread structure.
ASSERT(thread != nullptr);
thread->ResetHighWatermark();
// Set up other values and set the TLS value.
thread->isolate_ = nullptr;
thread->isolate_group_ = this;
thread->set_os_thread(os_thread);
ASSERT(thread->execution_state() == Thread::kThreadInNative);
thread->set_execution_state(Thread::kThreadInVM);
thread->set_safepoint_state(
Thread::SetBypassSafepoints(bypass_safepoint, 0));
thread->set_vm_tag(VMTag::kVMTagId);
ASSERT(thread->no_safepoint_scope_depth() == 0);
os_thread->set_thread(thread);
Thread::SetCurrent(thread);
os_thread->EnableThreadInterrupts();
}
return thread;
}最开始通过 OSThread::Current 取得当前线程，当前线程就是 ThreadPool 中执行 SpawnIsolateTask 的 worker 所在的线程，然后创建一个 Thread 实例，不过从 GetFreeThreadLocked 函数的实现来看，Thread 实例也有对应的复用机制，大概就是当一个线程销毁的时候 Thread 不会立刻销毁，而是被存到 free_list_ 中，而新线程开启，需要创建 Thread 时，便可以直接从这个变量中取出一个 Thread 实例，下面便是清空 Thread 中原有的数据，然后将其与 OS_Thread 建立联系，最后返回，传到上层之后与 Isolate 建立联系，至此，Isolate 的创建过程就结束了。
执行 Isolate
下一步就是回到 SpawnIsolateTask 的 Run 函数，在完成 Isoalte 的创建之后，就可以调用 Isolate 的 Run 函数开始运行了，准备开始执行一些业务层的逻辑代码。
// third_party/sdk/runtime/vm/isolate.cc
void Isolate::Run() {
message_handler()->Run(Dart::thread_pool(), RunIsolate, ShutdownIsolate,
reinterpret_cast<uword>(this));
}
// third_party/sdk/runtime/vm/message_handler.cc
void MessageHandler::Run(ThreadPool* pool,
StartCallback start_callback,
EndCallback end_callback,
CallbackData data) {
MonitorLocker ml(&monitor_);
if (FLAG_trace_isolates) {
OS::PrintErr(
"[+] Starting message handler:\n"
"\thandler: %s\n",
name());
}
ASSERT(pool_ == NULL);
ASSERT(!delete_me_);
pool_ = pool;
start_callback_ = start_callback;
end_callback_ = end_callback;
callback_data_ = data;
task_running_ = true;
const bool launched_successfully = pool_->Run<MessageHandlerTask>(this);
ASSERT(launched_successfully);
}从上面的函数实现可以看出，Isolate 会调用 MessageHandler 的 Run 函数，同时传入了 ThreadPool，RunIsolate、ShutDownIsolate 两个回调函数和 Isolate 自身，MessageHandler 保存了传递过来的值，随即创建出一个 MessageHandlerTask 实例，然后交由 ThreadPool 执行。此时代码运行在 ThreadPool 中的一个线程中，通过上面创建 Isoalte 过程可以看到，Isolate 会与这个线程建立联系，再根据对 Isolate 的理解，也就是 Isolate 的 MessageHandler 中会有一个无限循环，通过接收到的消息决定执行内容，所以 Isolate 所在线程对应的 worker 在 Isolate 存活的时间段中应该是不会再处于空闲状态的。那么按理说，MessageHandlerTask 直接在当前线程执行便好，为什么还要过一遍线程池？这样一来，就相当于 Isolate 同时运行在两个线程中了，且两个线程都在执行 HandleMessage 。
这一段不明了的先待定，下一步就是在线程池中找到一个空闲的 worker 执行 MessageHandlerTask，过程与执行 SpawnIsolateTask 如出一辙，最后会执行到 MessageHandlerTask Run 函数：
// third_party/sdk/runtime/vm/message_handler.cc
virtual void Run() {
ASSERT(handler_ != NULL);
handler_->TaskCallback();
}
void MessageHandler::TaskCallback() {
ASSERT(Isolate::Current() == NULL);
MessageStatus status = kOK;
bool run_end_callback = false;
bool delete_me = false;
EndCallback end_callback = NULL;
CallbackData callback_data = 0;
{
// We will occasionally release and reacquire this monitor in this
// function. Whenever we reacquire the monitor we *must* process
// all pending OOB messages, or we may miss a request for vm
// shutdown.
MonitorLocker ml(&monitor_);
// This method is running on the message handler task. Which means no
// other message handler tasks will be started until this one sets
// [task_running_] to false.
ASSERT(task_running_);
if (status == kOK) {
if (start_callback_ != nullptr) {
// Initialize the message handler by running its start function,
// if we have one. For an isolate, this will run the isolate's
// main() function.
//
// Release the monitor_ temporarily while we call the start callback.
ml.Exit();
status = start_callback_(callback_data_);
ASSERT(Isolate::Current() == NULL);
start_callback_ = NULL;
ml.Enter();
}
bool handle_messages = true;
while (handle_messages) {
handle_messages = false;
// Handle any pending messages for this message handler.
if (status != kShutdown) {
status = HandleMessages(&ml, (status == kOK), true);
}
if (status == kOK && HasLivePorts()) {
handle_messages = CheckIfIdleLocked(&ml);
}
}
}
// The isolate exits when it encounters an error or when it no
// longer has live ports.
if (status != kOK || !HasLivePorts()) {
if (FLAG_trace_isolates) {
if (status != kOK && thread() != NULL) {
const Error& error = Error::Handle(thread()->sticky_error());
OS::PrintErr(
"[-] Stopping message handler (%s):\n"
"\thandler: %s\n"
"\terror: %s\n",
MessageStatusString(status), name(), error.ToCString());
} else {
OS::PrintErr(
"[-] Stopping message handler (%s):\n"
"\thandler: %s\n",
MessageStatusString(status), name());
}
}
pool_ = NULL;
// Decide if we have a callback before releasing the monitor.
end_callback = end_callback_;
callback_data = callback_data_;
run_end_callback = end_callback_ != NULL;
delete_me = delete_me_;
}
// Clear task_running_ last. This allows other tasks to potentially start
// for this message handler.
ASSERT(oob_queue_->IsEmpty());
task_running_ = false;
}
// The handler may have been deleted by another thread here if it is a native
// message handler.
// Message handlers either use delete_me or end_callback but not both.
ASSERT(!delete_me || !run_end_callback);
if (run_end_callback) {
ASSERT(end_callback != NULL);
end_callback(callback_data);
// The handler may have been deleted after this point.
}
if (delete_me) {
delete this;
}
}首先调用 start_callback_ ，然后一个循环调用 HandleMessages 处理不断过来的消息，此时并没有处于 Isoalte 被创建的线程，start_callback_ 这个函数在 Isolate::Run 函数中赋值，也就是 RunIsolate 函数：
static MessageHandler::MessageStatus RunIsolate(uword parameter) {
Isolate* isolate = reinterpret_cast<Isolate*>(parameter);
IsolateSpawnState* state = nullptr;
{
// TODO(turnidge): Is this locking required here at all anymore?
MutexLocker ml(isolate->mutex());
state = isolate->spawn_state();
}
{
StartIsolateScope start_scope(isolate);
Thread* thread = Thread::Current();
ASSERT(thread->isolate() == isolate);
StackZone zone(thread);
HandleScope handle_scope(thread);
// If particular values were requested for this newly spawned isolate, then
// they are set here before the isolate starts executing user code.
isolate->SetErrorsFatal(state->errors_are_fatal());
if (state->on_exit_port() != ILLEGAL_PORT) {
const SendPort& listener =
SendPort::Handle(SendPort::New(state->on_exit_port()));
isolate->AddExitListener(listener, Instance::null_instance());
}
if (state->on_error_port() != ILLEGAL_PORT) {
const SendPort& listener =
SendPort::Handle(SendPort::New(state->on_error_port()));
isolate->AddErrorListener(listener);
}
// Switch back to spawning isolate.
if (!ClassFinalizer::ProcessPendingClasses()) {
// Error is in sticky error already.
#if defined(DEBUG)
const Error& error = Error::Handle(thread->sticky_error());
ASSERT(!error.IsUnwindError());
#endif
return MessageHandler::kError;
}
Object& result = Object::Handle();
result = state->ResolveFunction();
bool is_spawn_uri = state->is_spawn_uri();
if (result.IsError()) {
return StoreError(thread, Error::Cast(result));
}
ASSERT(result.IsFunction());
Function& func = Function::Handle(thread->zone());
func ^= result.raw();
func = func.ImplicitClosureFunction();
const Array& capabilities = Array::Handle(Array::New(2));
Capability& capability = Capability::Handle();
capability = Capability::New(isolate->pause_capability());
capabilities.SetAt(0, capability);
// Check whether this isolate should be started in paused state.
if (state->paused()) {
bool added = isolate->AddResumeCapability(capability);
ASSERT(added); // There should be no pending resume capabilities.
isolate->message_handler()->increment_paused();
}
capability = Capability::New(isolate->terminate_capability());
capabilities.SetAt(1, capability);
// Instead of directly invoking the entry point we call '_startIsolate' with
// the entry point as argument.
// Since this function ("RunIsolate") is used for both Isolate.spawn and
// Isolate.spawnUri we also send a boolean flag as argument so that the
// "_startIsolate" function can act corresponding to how the isolate was
// created.
const Array& args = Array::Handle(Array::New(7));
args.SetAt(0, SendPort::Handle(SendPort::New(state->parent_port())));
args.SetAt(1, Instance::Handle(func.ImplicitStaticClosure()));
args.SetAt(2, Instance::Handle(state->BuildArgs(thread)));
args.SetAt(3, Instance::Handle(state->BuildMessage(thread)));
args.SetAt(4, is_spawn_uri ? Bool::True() : Bool::False());
args.SetAt(5, ReceivePort::Handle(ReceivePort::New(
isolate->main_port(), true /* control port */)));
args.SetAt(6, capabilities);
const Library& lib = Library::Handle(Library::IsolateLibrary());
const String& entry_name = String::Handle(String::New("_startIsolate"));
const Function& entry_point =
Function::Handle(lib.LookupLocalFunction(entry_name));
ASSERT(entry_point.IsFunction() && !entry_point.IsNull());
result = DartEntry::InvokeFunction(entry_point, args);
if (result.IsError()) {
return StoreError(thread, Error::Cast(result));
}
}
return MessageHandler::kOK;
}在这里可以看到，首先通过 StartIsolateScope 使 Isoalte 与当前线程建立关联，这也就是为什么 ASSERT(thread->isolate() == isolate) 能够通过的原因，并且，在这个函数执行完了之后，这个类的析构函数会取消 Isolate 与线程的关联，使之满足 ASSERT(thread->isolate() == null)，由此可见，Isoalte 并不是与某一个线程绑定的，而是根据需要，可以在其他线程中与某一个 Isolate 进行绑定，然后解绑，但是在 flutter 中 Isoalte 是表示一个线程的，那么也就意味着，flutter 中认为的单线程 Isolate，实际上可能运行在不同线程中？接着就调用 spawn 函数时是否传入 onExit 和 onError 决定是否设置对应的监听器，然后调用到了 Dart 的 _startIsolate 函数，参数是 args 。
@pragma("vm:entry-point", "call")
void _startIsolate(
SendPort parentPort,
Function entryPoint,
List<String> args,
var message,
bool isSpawnUri,
RawReceivePort controlPort,
List capabilities) {
// The control port (aka the main isolate port) does not handle any messages.
if (controlPort != null) {
controlPort.handler = (_) {}; // Nobody home on the control port.
}
if (parentPort != null) {
// Build a message to our parent isolate providing access to the
// current isolate's control port and capabilities.
//
// TODO(floitsch): Send an error message if we can't find the entry point.
var readyMessage = new List(2);
readyMessage[0] = controlPort.sendPort;
readyMessage[1] = capabilities;
// Out of an excess of paranoia we clear the capabilities from the
// stack. Not really necessary.
capabilities = null;
parentPort.send(readyMessage);
}
assert(capabilities == null);
// Delay all user code handling to the next run of the message loop. This
// allows us to intercept certain conditions in the event dispatch, such as
// starting in paused state.
RawReceivePort port = new RawReceivePort();
port.handler = (_) {
port.close();
if (isSpawnUri) {
if (entryPoint is _BinaryFunction) {
(entryPoint as dynamic)(args, message);
} else if (entryPoint is _UnaryFunction) {
(entryPoint as dynamic)(args);
} else {
entryPoint();
}
} else {
entryPoint(message);
}
};
// Make sure the message handler is triggered.
port.sendPort.send(null);
}当 parentPort 不为 null 则向其发送一个消息告知 Isolate 已经创建并运行完成，然后本地又通过 RawReceivePort 延迟执行 entryPoint 函数，这一段代码其实与 Future 无二，只不过没有使用 Future 进行包装，使用了更底层的功能。
而此处的 parentPort 相关的逻辑，要从 Isolate.Spawn 函数中的 _spawnCommon 函数开始看。
static Future<Isolate> _spawnCommon(RawReceivePort readyPort) {
Completer completer = new Completer<Isolate>.sync();
readyPort.handler = (readyMessage) {
readyPort.close();
if (readyMessage is List && readyMessage.length == 2) {
SendPort controlPort = readyMessage[0];
List capabilities = readyMessage[1];
completer.complete(new Isolate(controlPort,
pauseCapability: capabilities[0],
terminateCapability: capabilities[1]));
} else if (readyMessage is String) {
// We encountered an error while starting the new isolate.
completer.completeError(new IsolateSpawnException(
'Unable to spawn isolate: ${readyMessage}'));
} else {
// This shouldn't happen.
completer.completeError(new IsolateSpawnException(
"Internal error: unexpected format for ready message: "
"'${readyMessage}'"));
}
};
return completer.future;
}此处传入的参数 readyPort 就是与上面 parentPort 对应的，由此可见，这里需要在 Isolate 创建完成之后获取到 Isolate 并将其包装成 Future 传出去，这也正是 Isolate.spawn 的返回值。
Isolate 间通信
上面是从调用 Isolate.spawn 到执行 entryPoint 的过程，仅止于开启一个 Isolate ，但是新的 Isolate 执行完之后数据的返回，还需要使用到 ReceivePort ，它底层的实现还是 RawReceivePort ，在此基础上又增加了一些操作，下面可以分别看一下 listene 和 send 函数都干了什么，在此之前先看下 ReceivePort 是什么，它与 RawReceivePort 有什么关系。
ReceivePort
// third_party/sdk/runtime/lib/isolate_patch.dart
@patch
factory ReceivePort() => new _ReceivePortImpl();
_ReceivePortImpl() : this.fromRawReceivePort(new RawReceivePort());
_ReceivePortImpl.fromRawReceivePort(this._rawPort) {
_controller = new StreamController(onCancel: close, sync: true);
_rawPort.handler = _controller.add;
}上面就是 ReceivePort 的构造函数调用过程，_ReceivePortImpl 中创建了一个 RawReceivePort 实例，将其作为参数调用 fromRawReceivePort，而在 fromRawReceivePort 中可以看到，先是创建了 StreamController 实例，然后将 RawReceivePort 的 handler 设置为 _controller.add，根据对 RawReceivePort 的了解，可以知道这里的 handler 就是接收到消息之后的回调函数，于是便可以将 ReceivePort 和 RawReceivePort 二者串联起来，ReceivePort 作为 RawReceivePort 业务层的一种实现而存在。
下面从 StreamController 的构造过程，看一下 StreamController 到底是什么。
factory StreamController(
{void onListen(),
void onPause(),
void onResume(),
onCancel(),
bool sync: false}) {
return sync
? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel)
: new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);
}StringController 的继承者分为同步和异步两种，这里的同步/异步是指发送者与接受者之间的关系，如果是同步，在某些情况下发送者需要等待接收者执行完了之后才能返回，而异步，则不会关心接收者是否执行完。默认是异步的，但此时传的却是要求同步。
class _AsyncStreamController<T> = _StreamController<T>
with _AsyncStreamControllerDispatch<T>;
class _SyncStreamController<T> = _StreamController<T>
with _SyncStreamControllerDispatch<T>;
abstract class _AsyncStreamControllerDispatch<T>
implements _StreamController<T> {
void _sendData(T data) {
_subscription._addPending(new _DelayedData<T>(data));
}
void _sendError(Object error, StackTrace stackTrace) {
_subscription._addPending(new _DelayedError(error, stackTrace));
}
void _sendDone() {
_subscription._addPending(const _DelayedDone());
}
}
abstract class _SyncStreamControllerDispatch<T>
implements _StreamController<T>, SynchronousStreamController<T> {
int get _state;
void set _state(int state);
void _sendData(T data) {
_subscription._add(data);
}
void _sendError(Object error, StackTrace stackTrace) {
_subscription._addError(error, stackTrace);
}
void _sendDone() {
_subscription._close();
}
}
_StreamController(this.onListen, this.onPause, this.onResume, this.onCancel);同步与异步之间的区别就在于这三个函数的实现，在异步 StreamController 中，接收到的消息都不会立刻执行，而是将其加入到待定区等待被执行，然后其他的就是继承自 _StreamController ，它的构造函数，也就是将传入的四个回调函数保存下来。
listen
ReceivePort 通过调用 listen 接收 SendPort 发送来的消息，这里再看下 listen 是如何实现的。
StreamSubscription listen(void onData(var message),
{Function onError, void onDone(), bool cancelOnError}) {
return _controller.stream.listen(onData,
onError: onError, onDone: onDone, cancelOnError: cancelOnError);
}直接将参数原封不同传给 _controller.stream.listen ，_controller 就是在构造函数中创建的 StreamController，它的实现是 _StreamController，在 _StreamController 中，
Stream<T> get stream => new _ControllerStream<T>(this);所以还需要看 _ControllerStream 是什么。
_ControllerStream(this._controller);它的构造函数很简单，就是将 _StreamController 保存起来，那么就直接看它的 listen 函数的实现：
StreamSubscription<T> listen(void onData(T data),
{Function onError, void onDone(), bool cancelOnError}) {
cancelOnError = identical(true, cancelOnError);
StreamSubscription<T> subscription =
_createSubscription(onData, onError, onDone, cancelOnError);
_onListen(subscription);
return subscription;
}调用了 _createSubscription 函数创建 StreamSubscription 实例并返回， _createSubscription 函数在 _StreamImpl 中有实现，但是在 _ControllerStream 中也有实现，所以这里真正调用的还是 _ControllerStream 中的实现：
StreamSubscription<T> _createSubscription(void onData(T data),
Function onError, void onDone(), bool cancelOnError) =>
_controller._subscribe(onData, onError, onDone, cancelOnError);这里调用了 _StreamController 的 _subscribe 函数，搞了半天又回到了 _StreamController 中。
StreamSubscription<T> _subscribe(void onData(T data), Function onError,
void onDone(), bool cancelOnError) {
if (!_isInitialState) {
throw new StateError("Stream has already been listened to.");
}
_ControllerSubscription<T> subscription = new _ControllerSubscription<T>(
this, onData, onError, onDone, cancelOnError);
_PendingEvents<T> pendingEvents = _pendingEvents;
_state |= _STATE_SUBSCRIBED;
if (_isAddingStream) {
_StreamControllerAddStreamState<T> addState = _varData;
addState.varData = subscription;
addState.resume();
} else {
_varData = subscription;
}
subscription._setPendingEvents(pendingEvents);
subscription._guardCallback(() {
_runGuarded(onListen);
});
return subscription;
}首先创建 _ControllerSubscription 实例，然后更新状态，然后将 _pendingEvents ，在 listen 调用之前已经发过来的消息交给 subscription ，由此，_StreamController 进入了订阅状态。
send
再看 send 过程，send 是 SendPort 的函数，SendPort 又是从 ReceivePort 中获取的，
SendPort get sendPort {
return _rawPort.sendPort;
}这里可以看出，ReceivePort 和 RawReceivePort 的 SendPort 是同一个，一般我们使用 SendPort 可以向 ReceivePort 发送消息，ReceivePort 则通过 listen 监听 SendPort 发送过来的消息，从 ReceivePort 的构造函数中可以看到，发送的消息最终会传递给到 StreamController 的 add 函数。
void add(T value) {
if (!_mayAddEvent) throw _badEventState();
_add(value);
}
void _add(T value) {
if (hasListener) {
_sendData(value);
} else if (_isInitialState) {
_ensurePendingEvents().add(new _DelayedData<T>(value));
}
}
bool get hasListener => (_state & _STATE_SUBSCRIBED) != 0;_add 是实现函数，它对 value 的处理分为两种情况，当已有 listener 的时候，也就是已经调用 listen 函数之后，会调用 _sendData 将其发送给订阅者，否则，就暂存在 _pendingEvents 中，在 listen 调用的时候一并传给订阅者。
void _sendData(T data) {
_subscription._add(data);
}
void _add(T data) {
assert(!_isClosed);
if (_isCanceled) return;
if (_canFire) {
_sendData(data);
} else {
_addPending(new _DelayedData<T>(data));
}
}
bool get _canFire => _state < _STATE_IN_CALLBACK;然后以 _SyncStreamControllerDispatch 为例看下 _sendData 的实现过程，同步状态下，当有一个消息来到时，先判断当前的状态，如果没有正在处理某一条消息，则直接调用 _sendData 发送消息，否则就将消息加入等待区。
void _addPending(_DelayedEvent event) {
_StreamImplEvents<T> pending = _pending;
if (_pending == null) {
pending = _pending = new _StreamImplEvents<T>();
}
pending.add(event);
if (!_hasPending) {
_state |= _STATE_HAS_PENDING;
if (!_isPaused) {
_pending.schedule(this);
}
}
}加入到 _pending 中后，会再根据状态决定是否调用 schedule 函数，即是否开始处理消息，处理延迟消息需要调用 schedule 函数。
void schedule(_EventDispatch<T> dispatch) {
if (isScheduled) return;
assert(!isEmpty);
if (_eventScheduled) {
assert(_state == _STATE_CANCELED);
_state = _STATE_SCHEDULED;
return;
}
scheduleMicrotask(() {
int oldState = _state;
_state = _STATE_UNSCHEDULED;
if (oldState == _STATE_CANCELED) return;
handleNext(dispatch);
});
_state = _STATE_SCHEDULED;
}
void handleNext(_EventDispatch<T> dispatch) {
assert(!isScheduled);
_DelayedEvent event = firstPendingEvent;
firstPendingEvent = event.next;
if (firstPendingEvent == null) {
lastPendingEvent = null;
}
event.perform(dispatch);
}
class _DelayedData<T> extends _DelayedEvent<T> {
final T value;
_DelayedData(this.value);
void perform(_EventDispatch<T> dispatch) {
dispatch._sendData(value);
}
}从上面可以看到，schedule 函数经过几次调用，还是会调用到 _sendData 函数，但与最开始直接调用这个函数不同的是，此时的调用是异步的，并不会阻塞 _add 函数的返回。