OpenHarmony—EventHandler源碼解析
EventHandler是用于處理線程間通信的一種機制，可以通過EventRunner創建新線程，將耗時的操作放到新線程上執行。這樣既不阻塞原來的線程，任務又可以得到合理的處理。比如：主線程使用EventHandler創建子線程，子線程做耗時的下載圖片操作，下載完成后，子線程通過EventHandler通知主線程，主線程再更新UI。
基本概念
EventRunner是一種事件循環器，循環處理從該EventRunner創建的新線程的事件隊列中獲取InnerEvent事件。InnerEvent是EventHandler投遞的事件。
EventHandler是一種用戶在當前線程上投遞InnerEvent事件到異步線程上處理的機制。每一個EventHandler和指定的EventRunner所創建的新線程綁定，并且該新線程內部有一個事件隊列。EventHandler可以投遞指定的InnerEvent事件到這個事件隊列。EventRunner從事件隊列里循環地取出事件，并在EventRunner所在線程執行processEvent回調。一般，EventHandler有兩個主要作用：
- 在不同線程間分發和處理InnerEvent事件。
- 延遲處理InnerEvent事件。
運作機制
使用EventHandler實現線程間通信的主要流程：
- EventHandler投遞具體的InnerEvent事件到EventRunner所創建的線程的事件隊列。
- EventRunner循環從事件隊列中獲取InnerEvent事件。
- 新線程上處理該事件：觸發InnerEvent的回調方法并觸發EventHandler的處理方法。
接口說明:
ohos.events.emitter(Emitter)
InnerEvent
進程內的事件。
EventData
發送事件時傳遞的數據。
EventPriority
用于表示事件被投遞的優先級。
代碼目錄：
foundation/
└──foundation/appexecfwk/standard
├── interfaces
│ ├── innerkits
│ | └── libeventhandler # 內部接口
│ └── napi
│ └── eventhandler # NAPI接口
└──— libs/libeventhandler # 機制實現代碼
類圖：
EventRunner:
EventRunner的create方法在創建Runner線程的同時會實例化一個EventQueue，并賦值給類內部的成員變量queue_ 。 EventRunner的Runner線程輪詢queue_，取出隊列中的InnerEvent事件，并處理該事件。
EventHandler：
EventHandler的各Send方法通過調用EventQueue的Insert方法將InnerEvent插入EventQueue的subEventQueues_中 或 idleEvents__ 中。
當調用Post方法，投遞某個回調函數時，該回調函數會賦值給InnerEvent的taskCallback_，并通過Send方法將將擁有回調函數的InnerEvent插入隊列。
其中，subEventQueues_是size為3的SubEventQueue的列表，3個SubEventQueue分別用于存儲三種不同優先級(IMMEDIATE/HIGH/LOW)的InnerEvent。
idleEvents_用于存儲優先級為IDLE的InnerEvent。
同理，Remove方法即從上述subEventQueue_ 或 idleEvent_ 中刪除某事件。
EventQueue
通過成員變量 ioWaiter_ 取出定時或延時的InnerEvent事件。同時實現對文件描述符監聽事件的回調。
ioWaiter_ 成員的類型默認使用NoneIoWaiter , 此時不支持文件描述符的監聽。隊列中的延時或定時事件的取出，通過NoneIoWaiter調用 std::condition_variable機制實現。
當EventHandler中的AddFileDescriptorListener被調用時，ioWaiter_成員的類型自動轉換成EpollIoWaiter類型。EpollIoWaiter支持文件描述符的監聽事件。此時，隊列中的延時或定時事件的取出，通過系統的epoll機制實現。
關鍵代碼摘錄：
EventRunner
// Start event looper.
for (auto event = queue_->GetEvent(); event; event = queue_->GetEvent()) { //監聽輪詢queue_
std::shared_ptr<EventHandler> handler = event->GetOwner();
// Make sure owner of the event exists.
if (handler) {
std::shared_ptr<Logger> logging = logger_;
std::stringstream address;
address << handler.get();
if (logging != nullptr) {
if (!event->HasTask()) {
logging->Log("Dispatching to handler event id = " + std::to_string(event->GetInnerEventId()));
} else {
logging->Log("Dispatching to handler event task name = " + event->GetTaskName());
}
}
handler->DistributeEvent(event); //執行aInnerEvent事件
if (logging != nullptr) {
logging->Log("Finished to handler(0x" + address.str() + ")");
}
}
// Release event manually, otherwise event will be released until next event coming.
event.reset();
}
EventQueue
void EventQueue::Insert(InnerEvent::Pointer &event, Priority priority) //事件或任務的插入
{
if (!event) {
HILOGE("Insert: Could not insert an invalid event");
return;
}
std::lock_guard<std::mutex> lock(queueLock_);
bool needNotify = false;
switch (priority) { //根據Event的優先級將該Event插入不同的事件隊列
case Priority::IMMEDIATE:
case Priority::HIGH:
case Priority::LOW: {
needNotify = (event->GetHandleTime() < wakeUpTime_);
InsertEventsLocked(subEventQueues_[static_cast<uint32_t>(priority)].queue, event);
break;
}
case Priority::IDLE: {
// Never wake up thread if insert an idle event.
InsertEventsLocked(idleEvents_, event);
break;
}
default:
break;
}
if (needNotify) {
ioWaiter_->NotifyOne();
}
}
InnerEvent::Pointer EventQueue::GetEvent() //事件或任務的取出
{
std::unique_lock<std::mutex> lock(queueLock_);
while (!finished_) {
InnerEvent::TimePoint nextWakeUpTime = InnerEvent::TimePoint::max();
InnerEvent::Pointer event = GetExpiredEventLocked(nextWakeUpTime);
if (event) {
return event;
}
WaitUntilLocked(nextWakeUpTime, lock);
}
HILOGD("GetEvent: Break out");
return InnerEvent::Pointer(nullptr, nullptr);
}
EventHandler
void EventHandler::DistributeEvent(const InnerEvent::Pointer &event) //執行InnerEvent事件
{
if (!event) {
HILOGE("DistributeEvent: Could not distribute an invalid event");
return;
}
// Save old event handler.
std::weak_ptr<EventHandler> oldHandler = currentEventHandler;
// Save current event handler into thread local data.
currentEventHandler = shared_from_this();
auto spanId = event->GetTraceId();
auto traceId = HiTrace::GetId();
bool allowTraceOutPut = AllowHiTraceOutPut(spanId, event->HasWaiter());
if (allowTraceOutPut) {
HiTrace::SetId(*spanId);
HiTracePointerOutPut(spanId, event, "Receive", HiTraceTracepointType::HITRACE_TP_SR);
}
InnerEvent::TimePoint nowStart = InnerEvent::Clock::now();
DeliveryTimeAction(event, nowStart);
if (event->HasTask()) { //如果有回調函數
// Call task callback directly if contains a task.
(event->GetTaskCallback())(); //執行回調函數
} else {
// Otherwise let developers to handle it.
ProcessEvent(event); //執行ProcessEvent
}
DistributeTimeAction(event, nowStart);
if (allowTraceOutPut) {
HiTrace::Tracepoint(HiTraceTracepointType::HITRACE_TP_SS, *spanId, "Event Distribute over");
HiTrace::ClearId();
if (traceId.IsValid()) {
HiTrace::SetId(traceId);
}
}
// Restore current event handler.
if (oldHandler.expired()) {
currentEventHandler = nullptr;
} else {
currentEventHandler = oldHandler;
}
}
