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 的延时任务处理机制有更深入的理解，并在自己的项目中借鉴其设计思路。