live555源码分析（二）基本组件上

布满荆棘的人生 2024-04-18 17:03 73阅读 0赞

**live555源码分析系列**

[live555源码分析（一）live555初体验][live555_live555]

[live555源码分析（二）基本组件上][live555]

[live555源码分析（三）基本组件下][live555 1]

[live555源码分析（四）RTSPServer分析][live555_RTSPServer]

[live555源码分析（五）DESCRIBE请求的处理][live555_DESCRIBE]

[live555源码分析（六）SETUP和PLAY请求的处理][live555_SETUP_PLAY]

[live555源码分析（七）播放过程][live555 2]

[live555源码分析（八）多播][live555 3]

# live555源码分析（二）基本组件上 #

### 文章目录 ###

*  live555源码分析（二）基本组件上
 *   *  一、整体框架
     *   *  1.1 live555的主要对象
         *  1.2 live555框架
     *  二、对象讲解
     *   *  2.1 TaskScheduler
         *   *  2.1.1 TaskScheduler介绍
             *  2.1.2 TaskScheduler的基本使用
             *  2.1.2 BasicTaskScheduler源码分析
         *  2.2 UsageEnvironment
         *  2.3 HashTable

在学习live555前，有必要先学习live555都有哪些对象，以及其框架

## 一、整体框架 ##

### 1.1 live555的主要对象 ###

live555包含许多对象，这里我将主要的类列举出来，关于每个对象，后面再进一步讲解

*  TaskScheduler
    
    任务调度器，负责live555的调度运行，live555是基于事件驱动的，要运行任务都需要向TaskScheduler添加
 *  UsageEnvironment
    
    环境变量，可以控制打印信息的输出，内部含有TaskScheduler的引用，在live555的大多数对象中传递
 *  HashTable
    
    哈希表，是live555中使用非常频繁的数据结构
 *  Groupsock
    
    实现了UDP单播和多播
 *  RTPInterface
    
    RTP接口，内含Groupsock，实现了UDP单播或多播，还有TCP单播
 *  RTCPInstance
    
    RTCP实例，内含RTPInterface作为其发送接口，此外还定义了一系列的RTCP协议的处理
 *  RTSPServer
    
    代表RTSP服务器
 *  RTSPClientConnection
    
    代表RTSP客户端的连接，定义了一系列的客户端请求处理
 *  RTSPClientSession
    
    代表RTSP客户端与服务器的连接会话，定义了一系列的客户端请求处理
 *  ServerMediaSession
    
    表示一个会话
 *  ServerMediaSubsession
    
    子会话，代表一路音频或者视频流，会话中可以有多个子会话
 *  FramedSource
    
    生产者，用于生产音视频数据（H.264或AAC等）
 *  RTPSink
    
    消费者，用于将音视频数据进行RTP打包，然后发送给客户端

### 1.2 live555框架 ###

live555各个对象之间的关系可以用下面的图总结

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg_size_16_color_FFFFFF_t_70]  
RtspServer包含`ServerMediaSession`在`rtsp://ip:port/session`中的session表示一个会话,服务器可以有多个`ServerMediaSession`

`ServerMediaSession`包含`ServerMediaSubSession`,一个`ServerMediaSubSession`表示一路音视频流,会话可以包含多个子会话

当客户端连接的时候,RtspServer会为其生成一个`RtspClientConnection`,`RtspClientConnection`处理客户端请求,当客户端发起`SETUP`请求的时候,会为其生成一个`RtspClientSession`

但客户端请求播放时,`RtspClientSession`会将客户端的信息加入`ServerMediaSubSession`,然后`RtpSink`开始播放,从`FramedSource`中获取音视频数据,然后RTP打包,通过`RTPInterface`发送,`RTPInterface`再通过`Groupsock`发送

每个`ServerMediaSubSession`中有一个`RTCPInstance`用于接收和发送RTCP信息

## 二、对象讲解 ##

### 2.1 TaskScheduler ###

#### 2.1.1 TaskScheduler介绍 ####

TaskScheduler作为live555的调度器，管理着live555的任务调度

其实TaskScheduler只是一个虚基类，定义了一系列的纯虚函数，给派生类具体实现

TaskScheduler定义的接口如下

class TaskScheduler {
        /* 添加定时事件 */
        virtual TaskToken scheduleDelayedTask(int64_t microseconds, TaskFunc* proc,
                                              void* clientData) = 0;
    
        #define SOCKET_READABLE    (1<<1)
    	#define SOCKET_WRITABLE    (1<<2)
    	#define SOCKET_EXCEPTION   (1<<3)
    	/* 添加套接字的监听事件 */
        virtual void setBackgroundHandling(int socketNum, int conditionSet, BackgroundHandlerProc* handlerProc, void* clientData) = 0;
    
    	/* 添加触发事件和触发事件 */
    	virtual EventTriggerId createEventTrigger(TaskFunc* eventHandlerProc) = 0;
    	virtual triggerEvent(EventTriggerId eventTriggerId, void* clientData = NULL) = 0;
        
        /* 事件循环处理函数 */
    	virtual void doEventLoop(char volatile* watchVariable = NULL) = 0;
    };

由此可看出，TaskScheduler要求其派生类实现三种事件，`定时事件`、`套接字监听事件`、`触发事件`，其中`套接字监听事件`支持三种监听方式，`SOCKET_READABLE（可读）`、`SOCKET_WRITABLE（可写）`、`SOCKET_EXCEPTION（异常）`

`doEventLoop`为事件循环处理函数，在主函数中，最后都会调用此函数进行事件循环处理

其实这就是一个reactor模式

live555实现了一个具体的TaskScheduler，其为`BasicTaskScheduler`，它是基于select实现的

继承关系如下

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg_size_16_color_FFFFFF_t_70 1]

其中`BasicTaskScheduler0`实现了定时器事件管理还有触发事件管理，`BasicTaskScheduler`基于select实现了套接字的监听事件管理，还有事件循环处理

#### 2.1.2 TaskScheduler的基本使用 ####

*  创建任务调度器
    
        TaskScheduler* scheduler = BasicTaskScheduler::createNew();

**定时事件**

*  添加定时任务
    
        void taskFunc(void* clientData)
        {
            int* data = (int*)clientData;
            //...
        }
        
        int data;
        TaskToken taskToken = scheduler->scheduleDelayedTask(microseconds, taskFunc, &data);
 *  注销定时任务
    
        scheduler->unscheduleDelayedTask(taskToken);

**触发事件**

*  创建触发事件
    
        void taskFunc(void* clientData)
        {
            int* data = (int*)clientData;
            //...
        }
        
        EventTriggerId eventTriggerId = scheduler->createEventTrigger(taskFunc);
 *  触发事件
    
        int data;
        
        scheduler->triggerEvent(eventTriggerId, &data);
 *  注销触发事件
    
        scheduler->deleteEventTrigger(eventTriggerId);

**监听事件**

*  添加监听事件
    
    触发条件为当套接字可读或者可写或者异常
    
        void taskFunc(void* clientData)
        {
              
            int* data = (int*)clientData;
            //...
        }
        
        int data;
        scheduler->setBackgroundHandling(socketNum, SOCKET_READABLE|SOCKET_WRITABLE|SOCKET_EXCEPTION, taskFunc, &data);
 *  注销监听事件
    
        scheduler->disableBackgroundHandling(socketNum);

**事件循环**

*  事件循环
    
    在主程序的最后一定要进入事件循环中
    
        scheduler->doEventLoop();

#### 2.1.2 BasicTaskScheduler源码分析 ####

我们先分析`BasicTaskScheduler0`如何管理定时事件和触发事件的,然后再分析`BasicTaskScheduler`如何使用select管理监听事件还有处理事件循环

首先查看`BasicTaskScheduler0`如何管理定时事件还有触发事件的,它定义了一个定时器队列还有一个32个元素的触发事件数组

class BasicTaskScheduler0 {
        ...
        
    private:
        DelayQueue fDelayQueue; // 定时器队列
        TaskFunc* fTriggeredEventHandlers[MAX_NUM_EVENT_TRIGGERS]; //32个元素的触发事件数组
    };

**定时事件**

看看其如何添加定时事件

TaskToken BasicTaskScheduler0::scheduleDelayedTask(int64_t microseconds,
    						 TaskFunc* proc,
    						 void* clientData)
    {
        DelayInterval timeToDelay((long)(microseconds/1000000), (long)(microseconds%1000000));
        AlarmHandler* alarmHandler = new AlarmHandler(proc, clientData, timeToDelay);
        
        fDelayQueue.addEntry(alarmHandler);
        return (void*)(alarmHandler->token());
    }

首先设置延迟时间,单位为微秒,然后生成一个定时器对象,添加到定时器队列中,然后再返回定时器的ID

`DelayQueue`是一个定时器队列,它使用双向链表实现一个队列,将定时器按超时时间从小到大进行排列,队列中的第一个定时器存放的是距离超时还有多长时间,之后的定时器存放的是跟前一个定时器的时间差

下面看一看添加定时器的函数

void DelayQueue::addEntry(DelayQueueEntry* newEntry) {
        /* 按照超时时间从小到大找到合适的位置 */
        DelayQueueEntry* cur = head();
        while (newEntry->fDeltaTimeRemaining >= cur->fDeltaTimeRemaining) {
            newEntry->fDeltaTimeRemaining -= cur->fDeltaTimeRemaining;
            cur = cur->fNext;
        }
        
        /* 添加到双向链表中 */
        newEntry->fNext = cur;
        newEntry->fPrev = cur->fPrev;
        cur->fPrev = newEntry->fPrev->fNext = newEntry;
    }

在事件循环中会调用定时器的处理函数,定时器处理函数会同步所有的定时器时间,然后处理超时的定时器

void DelayQueue::handleAlarm() {
        /* 同步定时器时间 */
    	if (head()->fDeltaTimeRemaining != DELAY_ZERO) synchronize();
    
        /* 处理超时的定时器 */
        if (head()->fDeltaTimeRemaining == DELAY_ZERO) {
        	DelayQueueEntry* toRemove = head();
        	removeEntry(toRemove);
            toRemove->handleTimeout();
        }
    }

看看`synchronize`是如何同步定时器超时时间的

void DelayQueue::synchronize() {
    	EventTime timeNow = TimeNow();
    
        /* 计算现在的时间和上一次同步时间的时间差 */
        DelayInterval timeSinceLastSync = timeNow - fLastSyncTime;
        
        /* 将所有超时的定时器的剩余时间设为0 */
        DelayQueueEntry* curEntry = head();
    	while (timeSinceLastSync >= curEntry->fDeltaTimeRemaining) {
            timeSinceLastSync -= curEntry->fDeltaTimeRemaining;
        	curEntry->fDeltaTimeRemaining = DELAY_ZERO;
            curEntry = curEntry->fNext;
        }
        
        /* 更新剩余定时器的超时时间 */
        curEntry->fDeltaTimeRemaining -= timeSinceLastSync;
    }

首先得到现在的时间和上一次同步时间的时间差,将所有超时的定时器的剩余时间设为0,重新更新剩余的定时器的超时时间

**触发事件**

查看如何创建触发事件

EventTriggerId BasicTaskScheduler0::createEventTrigger(TaskFunc* eventHandlerProc) {
    	do {
        	if (fTriggeredEventHandlers[i] == NULL) {
            	fTriggeredEventHandlers[i] = eventHandlerProc;
                return mask;
            }
        };
    }

从触发事件数组中找到空闲的触发事件,并返回其掩码(掩码是一个32bit的数,由于触发事件刚好也是32个元素,所以使用bitmap表示指定的触发事件)

下面看如何调用触发事件

void BasicTaskScheduler0::triggerEvent(EventTriggerId eventTriggerId, void* clientData) {
        /* 找到对应的触发事件 */
    	EventTriggerId mask = 0x80000000;
    	for (unsigned i = 0; i < MAX_NUM_EVENT_TRIGGERS; ++i) {
        	if ((eventTriggerId&mask) != 0)
        		fTriggeredEventClientDatas[i] = clientData;
        }
        
        /* 标记对应的触发事件 */
        fTriggersAwaitingHandling |= eventTriggerId;
    }

首先找到指定的触发事件,设置好触发事件的数据,然后在fTriggersAwaitingHandling中标记指定的触发事件(fTriggersAwaitingHandling是一个32bit的bitmap,用来指定哪一个触发事件被触发)

然后将在事件循环中根据标记,调用指定的触发事件

接下来分析`BasicTaskScheduler`如何管理套接字监听事件还有处理事件循环

**套接字监听事件**

`BasicTaskScheduler`是基于select实现的,其定义定义了三个监听集合

class BasicTaskScheduler: public BasicTaskScheduler0 {
        ...
    protected:
        fd_set fReadSet; //可读
        fd_set fWriteSet; //可写
        fd_set fExceptionSet; //异常
    };

看一下`BasicTaskScheduler`是如何添加监听事件的

void BasicTaskScheduler
      ::setBackgroundHandling(int socketNum, int conditionSet, BackgroundHandlerProc* handlerProc, 	void* clientData) {
      
      	/* 添加到链表中 */
    	fHandlers->assignHandler(socketNum, conditionSet, handlerProc, clientData);
    
    	/* 设置好监听集合 */
        if (conditionSet&SOCKET_READABLE) FD_SET((unsigned)socketNum, &fReadSet);
        if (conditionSet&SOCKET_WRITABLE) FD_SET((unsigned)socketNum, &fWriteSet);
        if (conditionSet&SOCKET_EXCEPTION) FD_SET((unsigned)socketNum, &fExceptionSet);
    }

首先将套接字监听事件添加到链表中,然后设置相应的集合

**事件循环处理**

接下来分析`BasicTaskScheduler`的事件循环处理,这也是整一个`TaskScheduler`的核心

void BasicTaskScheduler::SingleStep(unsigned maxDelayTime) {
        fd_set readSet = fReadSet; // make a copy for this select() call
        fd_set writeSet = fWriteSet; // ditto
        fd_set exceptionSet = fExceptionSet; // ditto
    
        /* 从定时器队列中获取最近的超时时间 */
        DelayInterval const& timeToDelay = fDelayQueue.timeToNextAlarm();
        
        /* 进入select阻塞等待 */
        int selectResult = select(fMaxNumSockets, &readSet, &writeSet, &exceptionSet, &tv_timeToDelay);
        
        /* 遍历套接字监听事件链表 */
        HandlerIterator iter(*fHandlers);
        while ((handler = iter.next()) != NULL) {
        	if (FD_ISSET(sock, &readSet) && FD_ISSET(sock, &fReadSet)
                resultConditionSet |= SOCKET_READABLE;
        	...
                
    		if ((resultConditionSet&handler->conditionSet) != 0 && handler->handlerProc != NULL) {
            	(*handler->handlerProc)(handler->clientData, resultConditionSet);
            	break;
            }
        }
             
    	/* 从bitmap中查找需要运行的触发事件 */
    	do {
         	if ((fTriggersAwaitingHandling&mask) != 0) {
             	fTriggersAwaitingHandling &=~ mask;
                (*fTriggeredEventHandlers[i])(fTriggeredEventClientDatas[i]);
                
                break;
            }
        } while (i != fLastUsedTriggerNum);
    
    	/* 处理超时的定时事件 */
    	fDelayQueue.handleAlarm();
    }

首先会获取定时器队列中最近的超时时间,将它作为select的超时时间,然后调用select进行多路IO监听

当select返回时,会先遍历套接字事件链表,找到其中第一个一个满足要求的事件运行,然后退出循环,其他事件等下一个事件循环处理

然后会检查bitmap,找到其中第一个需要执行的触发事件,执行它,然后退出循环,其他事件等下一个事件循环处理

最后处理定时器的一个超时事件

关于c就分析到这里

### 2.2 UsageEnvironment ###

`UsageEnvironment`是环境变量,控制打印输出,内含`TaskScheduler`的引用,在live555的大多数对象中传递

`UsageEnvironment`实际也是一个虚基类,其定义了一系列的虚函数,由派生类实现

class UsageEnvironment {
    public:
        TaskScheduler& taskScheduler() const {return fScheduler;}
        
        // 'console' output:
        virtual UsageEnvironment& operator<<(char const* str) = 0;
        virtual UsageEnvironment& operator<<(int i) = 0;
        virtual UsageEnvironment& operator<<(unsigned u) = 0;
        virtual UsageEnvironment& operator<<(double d) = 0;
        virtual UsageEnvironment& operator<<(void* p) = 0;
    
    private:
    	TaskScheduler& fScheduler;
    };

`BasicUsageEnvironment`是`UsageEnvironment`的具体实现,其实现了打印输出,它将内容输出到标准错误中

UsageEnvironment& BasicUsageEnvironment::operator<<(char const* str) {
        if (str == NULL) str = "(NULL)"; // sanity check
        fprintf(stderr, "%s", str);
        return *this;
    }

### 2.3 HashTable ###

`HashTable`表示哈希表,它其实是一个虚基类,定义了一系列的虚函数

class HashTable {
    	virtual void* Add(char const* key, void* value) = 0;
    	virtual Boolean Remove(char const* key) = 0;
        virtual void* Lookup(char const* key) const = 0;
    };

`BasicHashTable`是`HashTable`的一个派生类,它采用一种非常经典的方式实现哈希表

`BasicHashTable`使用一个bucket数组,每一个bucket维护着一个链表,如下图所示

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg_size_16_color_FFFFFF_t_70 2]

`BasicHashTable`中的元素

class TableEntry {
        
    public:
        TableEntry* fNext; //维护链表
        char const* key; //key值
        void* value; //value值
    };

`BasicHashTable`中的元素的key使用的是字符串,value使用void指针

查看`BasicHashTable`怎么添加元素

void* BasicHashTable::Add(char const* key, void* value) {
        void* oldValue;
        unsigned index;
        
        /* 查看该key是否存在哈希表中 */
    	TableEntry* entry = lookupKey(key, index);
    	
        if (entry != NULL) {
        	oldValue = entry->value;
        }
        else {
            entry = insertNewEntry(index, key);
            oldValue = NULL;
        }
        
        entry->value = value;
        
        if (fNumEntries >= fRebuildSize) rebuild();
        
        return oldValue;
    }

首先查看该key是否存在哈希表中,如果存在那么就只修改对应的value,如果不存在,那么新添加一个项

如果现在元素太多,桶太少,那么就进行扩容

看一看`lookupKey`,如何在哈希表中查找

BasicHashTable::TableEntry* BasicHashTable
    ::lookupKey(char const* key, unsigned& index) const {
    	index = hashIndexFromKey(key); //计算哈希值
    
    	for (entry = fBuckets[index]; entry != NULL; entry = entry->fNext) {
    		if (keyMatches(key, entry->key)) break;
    	}
    	
    	return entry;
    }

首先根据规定好的数学公式计算哈希值,然后找到指定的bucket,遍历该bucket的链表

看一下`insertNewEntry`如何插入新的项

BasicHashTable::TableEntry* BasicHashTable
    ::insertNewEntry(unsigned index, char const* key) {
      	TableEntry* entry = new TableEntry();
      	entry->fNext = fBuckets[index];
      	fBuckets[index] = entry;
    
      	++fNumEntries;
      	assignKey(entry, key);
    
      	return entry;
    }

首先找到指定的bucket,在bucket链表中添加项

`BasicHashTable`的扩容就是申请一个更大的数组,将原来老的哈希表项重新计算哈希值,插入到指定的bucket中,这里就不看代码了

这篇文章就到这里,关于其他对象的讲解,下一篇文章继续

[live555_live555]: https://blog.csdn.net/weixin_42462202/article/details/100626262
[live555]: https://blog.csdn.net/weixin_42462202/article/details/100677974
[live555 1]: https://blog.csdn.net/weixin_42462202/article/details/100709515
[live555_RTSPServer]: https://blog.csdn.net/weixin_42462202/article/details/100711104
[live555_DESCRIBE]: https://blog.csdn.net/weixin_42462202/article/details/100713429
[live555_SETUP_PLAY]: https://blog.csdn.net/weixin_42462202/article/details/100733866
[live555 2]: https://blog.csdn.net/weixin_42462202/article/details/100746999
[live555 3]: https://blog.csdn.net/weixin_42462202/article/details/100748292
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190909191839574.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20190909200050530.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20190909221548286.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ2MjIwMg==,size_16,color_FFFFFF,t_70