系列文章目录
第一篇 基于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之延时任务处理
文章目录
- 系列文章目录
- 前言
- 一、整体设计思路
- 二、源码分析
- 2.1 延时任务处理函数
- 2.2 getMinDelay 函数
- 2.3 flushDelayTask 函数
- 三、延时任务触发用户接口函数doDelayTask
- 四、设计亮点与启发
- 4.1 并发性能优化
- 4.2 可重复任务处理
- 4.3 异步插入任务
- 五、实际应用场景
- 5.1 定时任务
- 5.2 超时检测
- 总结
前言
前面两篇文章中已经讲到了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 的延时任务处理机制有更深入的理解，并在自己的项目中借鉴其设计思路。
