当前位置：首页 > news >正文

ZLMediaKit 源码分析——[4] ZLToolKit 中EventPoller之异步任务处理

news 来源：原创 2025/5/9 9:42:38

系列文章目录

第一篇基于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之异步任务处理

文章目录

系列文章目录
前言
一、异步任务处理原理
二、类图
三、源码分析
- 3.1添加管道监听事件
- 3.2 内部管道事件onPipeEvent
- 3.3 异步任务调度runLoop
四、异步事件接口
- 4.1 异步执行任务 async_l
- 4.2 异步执行任务 async
- 4.3 异步执行任务 async_first
五、使用方法
- 5.1 测试用例
- 输出如下： ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/42959c5cb15f4dad90ae562baa14450c.png)
总结

前言

在前面的文章中，我们详细剖析了 ZLToolKit 里 EventPoller 的网络事件处理机制。当深入研究 EventPoller 时，会发现它不仅负责网络事件的管理，还承担着异步任务处理的重要职责。这和 live555 里的设计类似，在事件循环中除了处理网络事件，也会执行异步任务和延时队列。今天，我们就来深入探讨一下 ZLToolKit 中 EventPoller 的异步任务处理机制。

一、异步任务处理原理

在 EventPoller 中，异步任务处理的核心在于将任务的执行与主线程分离，避免阻塞主线程的执行。当用户提交一个异步任务时，EventPoller 会判断是否适合同步执行，如果不适合同步执行，则将该任务添加到一个任务队列中。在 EventPoller 的事件循环中，会不断检查任务队列是否有任务需要执行，如果有则依次取出并执行这些任务。

具体来说，EventPoller 采用了生产者 - 消费者模型。用户代码作为生产者，将异步任务添加到任务队列；而 EventPoller 的事件循环则作为消费者，从任务队列中取出任务并执行。这种模型的好处是可以将耗时的任务异步执行，不影响主线程的正常运行，从而提高系统的并发处理能力。
其核心类为 EventPoller，关键函数是 runLoop ，异步事件处理中关键的数据结构为_list_task，用户接口为async和async_first 。

二、类图

这里对类图进行了简化，只抽取了和异步任务有关的类，即本文可能会讨论的地方。
在这里插入图片描述

三、源码分析

3.1添加管道监听事件

再看addEvent函数

int EventPoller::addEvent(int fd, int event, PollEventCB cb) {// 时间检查TimeTicker();// 回调函数检查if (!cb) {WarnL << "PollEventCB is empty";return -1;}// 当前线程检查，// 如果是当前线程：根据不同的操作系统平台选择不同的事件通知机制来添加事件监听。if (isCurrentThread()) {
#if defined(HAS_EPOLL)struct epoll_event ev = {0};ev.events = toEpoll(event) ;ev.data.fd = fd;int ret = epoll_ctl(_event_fd, EPOLL_CTL_ADD, fd, &ev);if (ret != -1) {_event_map.emplace(fd, std::make_shared<PollEventCB>(std::move(cb)));}return ret;
#elif defined(HAS_KQUEUE)
#else
#endif}// 如果不是当前线程，发起异步操作：// 使用 async 函数异步调用 addEvent 函数，将事件添加操作放到合适的线程中执行。async([this, fd, event, cb]() mutable {addEvent(fd, event, std::move(cb));});return 0;
}

再看函数后面非当前线程处理这里，如果不是当前线程，这里会调用async函数，将会再次addEvent函数，直到是当前线程，添加进_event_map中，对这里的理解很关键，结合下面这个函数，可以带着几个问题，addEvent事件会调用多次吗？onPipeEvent是什么时候调用的？onPipeEvent会调用多次吗？理解了这里，就基本上理解了ZLMediaKit中异步任务的关键。

void EventPoller::addEventPipe() {SockUtil::setNoBlocked(_pipe.readFD());SockUtil::setNoBlocked(_pipe.writeFD());// 添加内部管道事件if (addEvent(_pipe.readFD(), EventPoller::Event_Read, [this](int event) { onPipeEvent(); }) == -1) {throw std::runtime_error("Add pipe fd to poller failed");}
}

EventPoller::addEventPipe() 函数的主要功能是对内部管道的读写文件描述符进行非阻塞设置，并将管道的读文件描述符添加到事件轮询器中，以便监听管道的可读事件。当管道有可读事件发生时，会调用 onPipeEvent() 函数进行相应处理。

3.2 内部管道事件onPipeEvent

inline void EventPoller::onPipeEvent(bool flush) {char buf[1024];int err = 0;if (!flush) {for (;;) {if ((err = _pipe.read(buf, sizeof(buf))) > 0) {// 读到管道数据,继续读,直到读空为止continue;}if (err == 0 || get_uv_error(true) != UV_EAGAIN) {// 收到eof或非EAGAIN(无更多数据)错误,说明管道无效了,重新打开管道ErrorL << "Invalid pipe fd of event poller, reopen it";delEvent(_pipe.readFD());_pipe.reOpen();addEventPipe();}break;}}decltype(_list_task) _list_swap;{lock_guard<mutex> lck(_mtx_task);_list_swap.swap(_list_task);}_list_swap.for_each([&](const Task::Ptr &task) {try {(*task)();} catch (ExitException &) {_exit_flag = true;} catch (std::exception &ex) {ErrorL << "Exception occurred when do async task: " << ex.what();}});
}

EventPoller::onPipeEvent 函数的核心功能是处理管道事件。它会根据传入的 flush 参数决定是否清空管道，在管道出现异常时进行重新打开操作，同时会执行之前存储在任务列表中的异步任务。

函数原型void onPipeEvent(bool flush = false);参数flush 默认值为false，整个代码中只有两处调用onPipeEvent，一处为1中添加管道监听事件中addEventPipe()中，传参为false。
第二处为析构函数中，传参为true。

EventPoller::~EventPoller() {//...其它代码onPipeEvent(true);InfoL << getThreadName();
}

3.3 异步任务调度runLoop

EventPoller::runLoop 函数是 EventPoller 类的核心函数，它实现了事件循环的逻辑，不断地监听网络套接字和内部管道的事件并进行处理。

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;}_event_cache_expired.clear();for (int i = 0; i < ret; ++i) {struct epoll_event &ev = events[i];int fd = ev.data.fd;if (_event_cache_expired.count(fd)) {//event cache refreshcontinue;}auto it = _event_map.find(fd);if (it == _event_map.end()) {epoll_ctl(_event_fd, EPOLL_CTL_DEL, fd, nullptr);continue;}auto cb = it->second;try {(*cb)(toPoller(ev.events)); // 将epoll事件类型转换回自定事件类型} catch (std::exception &ex) {ErrorL << "Exception occurred when do event task: " << ex.what();}}}
#elif defined(HAS_KQUEUE)// ... 其他系统的处理代码
#else// ... 其他系统的处理代码
#endif //HAS_EPOLL} else {_loop_thread = new thread(&EventPoller::runLoop, this, true, ref_self);_sem_run_started.wait();}
}

这个函数的处理逻辑在上篇博文中已经详细介绍了，这里与网络事件区别是监听的是无名管道读端描述符，addEventPipe中通过监听addEvent(_pipe.readFD(), EventPoller::Event_Read, [this](int event) { onPipeEvent(); })加入的_pipe.readFD()读端管道描述符，这里执行的cb即为[this](int event) { onPipeEvent(); }。

四、异步事件接口

4.1 异步执行任务 async_l

源码如下：

Task::Ptr EventPoller::async_l(TaskIn task, bool may_sync, bool first) {TimeTicker();if (may_sync && isCurrentThread()) {task();return nullptr;}auto ret = std::make_shared<Task>(std::move(task));{lock_guard<mutex> lck(_mtx_task);if (first) {_list_task.emplace_front(ret);} else {_list_task.emplace_back(ret);}}//写数据到管道,唤醒主线程_pipe.write("", 1);return ret;
}

EventPoller::async_l 函数的主要功能是根据传入的参数决定是同步执行任务还是异步执行任务。如果可以同步执行，则直接执行任务；否则，将任务封装到 Task 对象中，并插入到任务列表中，然后通过向管道写入数据唤醒主线程来处理任务。最后返回异步任务的智能指针。从这里也能看出保存异步任务列表的关键的数据结构是_list_task，原型为ListTask::Ptr _list_task。

4.2 异步执行任务 async

对async_l的封装，将任务加到队尾执行

Task::Ptr EventPoller::async(TaskIn task, bool may_sync) {return async_l(std::move(task), may_sync, false);
}

4.3 异步执行任务 async_first

对async_l的封装，将任务加到队列头部，表示插入优先级高的任务

Task::Ptr EventPoller::async_first(TaskIn task, bool may_sync) {return async_l(std::move(task), may_sync, true);
}

五、使用方法

5.1 测试用例

#include "EventPoller/EventPoller.h"
#include <iostream>
using namespace toolkit;int main() 
{for (int i = 0; i < 10; i++) {EventPollerPool::Instance().getPoller()->async([i]{ std::cout << "hello task " << i << std::endl;std::this_thread::sleep_for(std::chrono::seconds(1));});}std::this_thread::sleep_for(std::chrono::seconds(10));return 0;
}

输出如下：

总结

本文深入剖析了 ZLToolKit 中 EventPoller 的异步任务处理机制。详细讲解了其核心原理，即采用生产者 - 消费者模型将任务执行与主线程分离，提高系统并发处理能力。通过对关键类 EventPoller、函数 runLoop、数据结构 _list_task 以及用户接口 async 和 async_first 的分析，展示了异步任务处理的实现方式。

在源码层面，addEvent 函数会根据线程情况决定同步或异步添加事件监听；addEventPipe 函数对内部管道进行非阻塞设置并监听读事件；onPipeEvent 函数处理管道事件并执行异步任务；runLoop 函数实现事件循环，监听管道读端描述符。同时，还介绍了 async_l、async 和 async_first 等异步事件接口的功能，最后还给了一个简单的测试用例。

通过对这些机制的理解，开发者能够更好地利用 ZLToolKit 进行网络编程，实现高效的异步任务处理。后续可能还会对相关内容进行进一步分析，敬请期待。