ZLMediaKit中的线程

news2024/9/30 17:39:38

EventLoop的线程模型

服务器通用的IO模型event-loop + 非阻塞IO。线程模型可以是单线程，可以是多线程。对于已经普及了的多核环境，通常都是采用多线程。

通常一个线程中有一个EventLoop，比如accept是一个专门线程，accept后的fd分到不同线程中，一个fd只会在一个EventLoop中处理。那么也可以说是一个EventLoop就代表了一个线程。在ZLMediaKit中线程就是这种抽象。

EventLoop的基本实现就是IO多路复用(select，epoll等)+消息队列，它不止可以用于处理IO。也可以当作消息队列或定时器使用。

这种方式非常适用于对线程职责的划分，将特定的业务通过消息队列分到固定的线程。在webrtc中，rtc::thread也是种实现。它同时作为网络线程，信令线程，工作线程使用。其中工作线程并没有IO功能。 详细见webrtc中的rtc::thread

ZLMediaKit中的EventLoop

线程

EventLoop在众多的开源网络库都有实现。ZLMediaKit的实现非常适合用于多线程模型。

企业微信截图_17014217474387.png

以上为其类图，EventPoller就是EventLoop，为一个事件循环。它继承自TaskExecutor，代表了一个线程。

其中的方法async和sync分别代表着向线程投递异步任务和同步任务(在TaskExecutorInterface中定义)。它还包括了计算CPU负载的功能(在ThreadLoadCounter中定义)。

所以EventPoller是一个功能相当完备的线程：

有事件循环
可以向其投递任务。
可以计算CPU负载，结合线程池一起使用，选取负载低的线程。

线程池

企业微信截图_1701421705841.png

以上为ZLMediaKit中线程池类图，通过EventPollerPooll的getPoller或getFirstPoller方法来取一个线程(EventPoller)。getPoller优先取一个CPU负载小的线程。

使用

产生一个线程

在ZLMediaKit中通过EventPollerPool来获取一个EventPoller，如下代码:

EventPollerPool::setPoolSize(thread_num);
auto poller = EventPollerPool::Instance().getPoller();

EventPollerPool::setPoolSize(thread_num)设置线程数，当产生线程池对象EventPollerPool时，线程都已创建并绑定到了核上，通过getPoller()方法获取一个CPU负责最小的线程。上面的poller对象就代表了一个线程。

向线程投递一个任务

通过async，比如如下代码:

poller->async([invoker,contentOut](){
                HttpSession::KeyValue headerOut;
                //你可以自定义header,如果跟默认header重名，则会覆盖之
                //默认header有:Server,Connection,Date,Content-Type,Content-Length
                //请勿覆盖Connection、Content-Length键
                //键名覆盖时不区分大小写
                headerOut["TestHeader"] = "HeaderValue";
                invoker(200,headerOut,contentOut);
            });

通过doDelayTask向线程投递定时任务，如下代码:

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);
        }
    });