ZLMediaKit 源码分析——[5] ZLToolKit 中EventPoller之延时任务处理

系列文章目录

第一篇 基于SRS 的 WebRTC 环境搭建
第二篇 基于SRS 实现RTSP接入与WebRTC播放
第三篇 centos下基于ZLMediaKit 的WebRTC 环境搭建
第四篇 WebRTC学习一：获取音频和视频设备
第五篇 WebRTC学习二：WebRTC音视频数据采集
第六篇 WebRTC学习三：WebRTC音视频约束
第七篇 WebRTC学习四：WebRTC常规视觉滤镜
第八篇 WebRTC学习五：从视频中提取图片
第九篇 WebRTC学习六：MediaStream 常用API介绍
第十篇 WebRTC学习七：WebRTC 中 STUN 协议详解
ZLMediaKit源码分析——[1] 开篇：基础库 ZLToolKit 之 onceToken 源码分析
ZLMediaKit 源码分析——[2] 从 ZLToolKit 代码看 CPU 亲和性设计
ZLMediaKit 源码分析——[3] ZLToolKit 中EventPoller之网络事件处理
ZLMediaKit 源码分析——[4] ZLToolKit 中EventPoller之异步任务处理
ZLMediaKit 源码分析——[5] ZLToolKit 中EventPoller之延时任务处理

---

---

前言

前面两篇文章中已经讲到了EventPoller中的网络事件处理机制和异步任务处理机制，有了前面两篇文章的IO框架和异步任务的基础，今天的延时任务的内容就比较容易理解了，今天我们就一鼓作气，把ZLMediaKit中延时任务的机制和实现细节分析一下，结束掉EventPoller的分析。

---

一、整体设计思路

EventPoller 中的延时任务处理主要基于一个有序的任务队列 _delay_task_map。这个队列以任务的执行时间戳作为键，以任务对象作为值，按照时间戳从小到大的顺序排列。通过这种方式，我们可以很方便地找到最近需要执行的任务。

在处理延时任务时，主要涉及三个关键函数：flushDelayTask、getMinDelay 和 doDelayTask以及数据结构std::multimap<uint64_t, DelayTask::Ptr> _delay_task_map，下面我们将分别对这三个函数和_delay_task_map进行详细分析。

二、源码分析

2.1 延时任务处理函数

void EventPoller::runLoop(bool blocked, bool ref_self) {if (blocked) {if (ref_self) {s_current_poller = shared_from_this();}_sem_run_started.post();_exit_flag = false;uint64_t minDelay;
#if defined(HAS_EPOLL)struct epoll_event events[EPOLL_SIZE];while (!_exit_flag) {minDelay = getMinDelay();startSleep();//用于统计当前线程负载情况int ret = epoll_wait(_event_fd, events, EPOLL_SIZE, minDelay ? minDelay : -1);sleepWakeUp();//用于统计当前线程负载情况if (ret <= 0) {//超时或被打断continue;}// ...其它处理代码}
#elif defined(HAS_KQUEUE)// ... 其他系统的处理代码
#else// ... 其他系统的处理代码
#endif //HAS_EPOLL} else {_loop_thread = new thread(&EventPoller::runLoop, this, true, ref_self);_sem_run_started.wait();}
}

与live555中在最后处理延时队列中的任务不一样，在ZLToolKit中会优先处理延时队列任务，在消息循环函数中会调用getMinDelay()，所有延时队列任务的处理将在这个函数里完成。

2.2 getMinDelay 函数

uint64_t EventPoller::getMinDelay() {// 查找最早延时任务auto it = _delay_task_map.begin();if (it == _delay_task_map.end()) {//没有剩余的定时器了return 0;}// 获取当前时间auto now = getCurrentMillisecond();// 如果最早任务的执行时间大于当前时间，说明所有任务都尚未到期，函数返回最早任务执行时间与当前时间的差值，即需要等待的时间。if (it->first > now) {//所有任务尚未到期return it->first - now; // 计算需要等待的时间}//执行已到期的任务并刷新休眠延时return flushDelayTask(now);
}

这里首先获取_delay_task_map 的首元素，_dealy_task_map定义是std::multimap<uint64_t, DelayTask::Ptr> _delay_task_map;‌ std::multimap是C++标准库中的一个容器，它存储的元素都是键值对，并且允许有重复的键‌。这与std::map不同，后者不允许有重复的键。那么这个函数的意义就很明确了：
1、查找最早延时任务，没有返回0；
2、如果所有任务都尚未到期，函数返回需要等待的时间；
3、执行已到期的任务并刷新休眠延时，返回下一个未到期任务执行时所需的最小延时时间。
我们接下来看下执行已到期的任务并刷新休眠延时 flushDelayTask(now)函数里面做了些什么事情。

2.3 flushDelayTask 函数

uint64_t EventPoller::flushDelayTask(uint64_t now_time) {// 交换任务队列，将当前任务转移到临时容器，避免处理期间阻塞新任务写入。decltype(_delay_task_map) task_copy; // 自动推导_delay_task_map类型，task_copy初始化为空task_copy.swap(_delay_task_map); // 交换 task_copy 和 _delay_task_map 的内容。task_copy 持有原来的 _delay_task_map 数据。_delay_task_map 变为空容器。// 处理到期任务‌：执行所有时间戳 ≤ now_time 的任务，并根据返回值决定是否重新调度。for (auto it = task_copy.begin(); it != task_copy.end() && it->first <= now_time; it = task_copy.erase(it)) {//已到期的任务try {auto next_delay = (*(it->second))();if (next_delay) {//可重复任务,更新时间截止线_delay_task_map.emplace(next_delay + now_time, std::move(it->second));}} catch (std::exception &ex) {ErrorL << "Exception occurred when do delay task: " << ex.what();}}// 合并新任务和未处理任务‌，确保未到期任务和新添加的重复任务保留在队列中。task_copy.insert(_delay_task_map.begin(), _delay_task_map.end());task_copy.swap(_delay_task_map);auto it = _delay_task_map.begin();if (it == _delay_task_map.end()) {//没有剩余的定时器了return 0;}//最近一个定时器的执行延时return it->first - now_time;
}

EventPoller::flushDelayTask(uint64_t now_time) 函数的主要功能是处理所有已到期的延时任务，并根据任务的返回值决定是否重新调度这些任务，最后返回距离下一个未到期任务执行所需的最小延时时间。

三、延时任务触发用户接口函数doDelayTask

延时任务的外部调用接口为doDelayTask, 所有想要添加一个延时任务，需要通过调用doDelayTask添加，传入第一个参数为需要延时的毫秒值，第二个参数为task 任务，task任务返回值为0时代表不再重复任务，否则为下次执行延时，如果任务中抛异常，那么默认不重复任务。doDelayTask的返回值为DelayTask::Ptr，定义为 using DelayTask = TaskCancelableImp<uint64_t(void)>;是一个可取消的任务，意味着在还没到任务执行之前是可以取消的。

EventPoller::DelayTask::Ptr EventPoller::doDelayTask(uint64_t delay_ms, function<uint64_t()> task) {DelayTask::Ptr ret = std::make_shared<DelayTask>(std::move(task));auto time_line = getCurrentMillisecond() + delay_ms; // 当前时间+需要延时执行的时间async_first([time_line, ret, this]() {//异步执行的目的是刷新select或epoll的休眠时间_delay_task_map.emplace(time_line, ret);});return ret;
}

该函数的主要功能是创建一个延时任务，将其添加到延时任务映射_delay_task_map中，该任务将在指定的延时时间（delay_ms)后执行。

四、设计亮点与启发

4.1 并发性能优化

通过使用任务队列交换的方式，避免了在处理任务时阻塞新任务的添加，提高了程序的并发性能。这种设计思路在处理高并发场景下的任务调度非常有效。

4.2 可重复任务处理

支持可重复执行的任务，通过任务返回的延时时间来重新调度任务，增加了任务处理的灵活性。这种设计使得程序可以方便地实现定时任务的功能。

4.3 异步插入任务

使用异步方式插入任务，确保了系统能够及时响应新的延时任务。这在网络编程中尤为重要，因为网络事件的处理需要高实时性。

五、实际应用场景

5.1 定时任务

可以使用 doDelayTask 函数来实现定时任务，例如定时清理缓存、定时发送心跳包等。
如zlToolKit中定时器的实现

Timer::Timer(float second, const std::function<bool()> &cb, const EventPoller::Ptr &poller) {_poller = poller;if (!_poller) {_poller = EventPollerPool::Instance().getPoller();}_tag = _poller->doDelayTask((uint64_t) (second * 1000), [cb, second]() {try {if (cb()) {//重复的任务return (uint64_t) (1000 * second);}//该任务不再重复return (uint64_t) 0;} catch (std::exception &ex) {ErrorL << "Exception occurred when do timer task: " << ex.what();return (uint64_t) (1000 * second);}});
}

这里Timer构造函数通过事件轮询器构建了一个定时任务，该任务会在指定的时间间隔后执行一个回调函数。根据回调函数的返回值，定时器可以选择重复执行或停止执行。同时，代码中对回调函数的异常进行了捕获和处理，确保在出现异常时定时器仍然可以继续运行。

5.2 超时检测

在网络通信中，经常需要检测连接是否超时。可以通过设置延时任务来实现超时检测，当任务到期时，如果连接还没有收到响应，则认为连接超时。
比如MediaSource.cpp 的findAsync_l函数中代码：

auto on_timeout = poller->doDelayTask(maxWaitMS, [cb_once, listener_tag]() {// 最多等待一定时间，如在这个时间内，流还未注册上，则返回空NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged);cb_once(nullptr);return 0;});

这里如果在最大等待时间里没有找到媒体源，将触发超时回调。

---

总结

ZLToolKit 中的 EventPoller 提供了一套高效、灵活的延时任务处理机制。通过有序的任务队列、任务队列交换、异步插入任务等技术手段，保证了程序的高并发性能和实时性。在实际应用中，我们可以根据具体需求灵活运用这些功能，实现各种复杂的定时任务和超时检测功能。

希望通过本文的分析，大家能够对 ZLMediaKit 中 EventPoller 的延时任务处理机制有更深入的理解，并在自己的项目中借鉴其设计思路。