[C++/流媒体/tcp/rtsp]构建一个简单的流媒体转发服务器，用于将rtsp推流转发出去

2026-01-28 18:52:18 栏目：最新资讯 2 阅读

使用效果举例:

运行程序代码后，打开cmd，执行以下代码进行桌面推流:
ffmpeg -f gdigrab -i desktop -c:v libx264 -f rtsp -rtsp_transport tcp rtsp://127.0.0.1/live/test

这会将流推送给流媒体，创建了一条流媒体的转发流rtsp://127.0.0.1/live/test。
随后使用potplayer或者vlcplayer拉取rtsp://127.0.0.1/live/test这条流，即可成功播放转发流。

注意，这个流媒体是简单的rtsp流媒体转发程序，不做rtp和rtcp包的分析和重构处理，仅原封不动地将发送至流媒体的rtsp流转发出去。

开篇:

作为流媒体，他的并发能力很重要，因此从搭建之初我就引入了wepoll库（很小，就两个文件），以其为核心做开发。

最初项目结构为

main.cpp 主程序
wepoll.h 三方库
wepoll.c 三方库

随后开始扩展

过程:

过程1:对wepoll进行封装，做套接字的并发处理，从而产生一个新类EventPoller。
过程2:建立RtspServer类，接入EventPoller的并发循环，开始accpet客户端。
过程3:得到推流到RtspServer的客户端连接，将其命名为RtspSession类使用。
过程4:新建SdpParser类来解析推流的SDP交互报文，确定流的通道参数，给RtspSession分析流通道的数据。
过程5:得到向RtspServer申请拉流的客户端连接，同样是RtspSession类实例。
过程6:新建MediaPusherManager类，用于追踪管理和区分推/拉流的RtspSession。

总项目结构完成，5个类完成一个简单的流媒体转发服务器。

main.cpp

wepoll.h
wepoll.c

EventPoller.h
EventPoller.cpp

RtspServer.h
RtspServer.cpp

RtspSession.h
RtspSession.cpp

SdpParser.h
SdpParser.cpp

MediaPusherManager.h
MediaPusherManager.cpp

代码:

具体的代码这里暂不展开讲解，已上传至git可供下载，支持直接编译运行

git项目:https://github.com/MingTianDay9/RtspServer

如果无法登录git进行克隆，那么下面是可以直接复制学习的代码文件
(没有wepoll，这拥有两个文件的库需自行搜索，git是https://github.com/piscisaureus/wepoll)

1.main.cpp

#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include 
#include 
#include 
#pragma comment (lib, "Ws2_32.lib")
#pragma comment(lib,"Iphlpapi.lib")
#include 
#include 
#include 
#include 

#include "EventPoller.h"
#include "RtspServer.h"

int main()
{
	bool isRet;
	//检查网络WSA是否可用，不可用则退出
	WORD wVersionRequested = MAKEWORD(2, 2);
	WSADATA wsaData;
	isRet = (0 == WSAStartup(wVersionRequested, &wsaData));
	if(false == isRet)
		return 1;
	EventPoller eventPoller;
	auto rtspServer = std::make_shared<RtspServer>(&eventPoller);
	if (rtspServer->Start("127.0.0.1", 554)) {
		std::cout << "已就绪，请试着用ffmpeg运行如下命令进行推流:ffmpeg -f gdigrab -i desktop -c:v libx264 -f  rtsp -rtsp_transport tcp rtsp://127.0.0.1/live/test
";
		std::cout << "推流成功后，可以使用某个播放器(比如VlcPlayer/PotPlayer)对rtsp://127.0.0.1/live/test进行拉流以查看转发效果
";
		eventPoller.Exec();
		rtspServer->Stop();
	}
	rtspServer.reset();

	WSACleanup();
	return 0;
}

2.EventPoller.h

#pragma once

#include 
#include 
#include 

typedef void* HANDLE;

/// 
/// epoll处理线程:这是对epoll三方库的进一步封装，它为网络套接字服务，负责单线程内处理千万级以上的套接字连接
/// 每个类实例都封装了一条线程用于接收并处理套接字的相关业务
/// 
class EventPoller {
public:
    using PollEventCB = std::function<void(int event)>;         //套接字的监听回调

    //提供给wepoll的监听事件类型
    typedef enum {
        Event_Read = 1 << 0,    // 读事件，对标wepoll.h的EPOLLIN
        Event_Write = 1 << 2,   // 写事件，对标wepoll.h的EPOLLOUT
        Event_Error = 1 << 3,   // 错误事件，对标wepoll.h的EPOLLERR
    } Poll_Event;

    /// 
    /// 构造
    /// 
    EventPoller();
    ~EventPoller();

    /// 
    /// 添加事件监听
    /// 
    /// 监听的套接字
    /// 事件类型，例如 Event_Read | Event_Write
    /// 事件回调functional
    /// 是否成功
    bool AddEvent(int fd, int event, PollEventCB cb);

    /// 
    /// 删除事件监听
    /// 
    /// 监听的套接字
    /// 是否成功
    bool DelEvent(int fd);

    /// 
    /// 执行事件轮询死循环
    /// 
    void Exec();

    /// 
    /// 获取当前线程下所有socket共享的读缓存
    /// 
    /// 
    std::vector<char>& GetSharedBuffer();
private:
    bool m_isRun = false;               //处理线程是否正在执行
    HANDLE m_hEpoll = NULL;             //本例中管控socket的epoll的fd
    std::unordered_map<int, std::shared_ptr<PollEventCB>> m_mapEventCB; //监听的套接字和他的处理回调 <套接字，处理回调>

    std::vector<char> m_vecSharedBuffer;            //当前线程下所有socket共享的读缓存
};

3.EventPoller.cpp

#include "EventPoller.h"

#include "wepoll.h"
#include "Windows.h"

//可管理的最大套接字数量，对标select的最大管理数量，但其实这个值并没有用，传进epoll_create里会被忽略
#define EPOLL_SIZE 1024

EventPoller::EventPoller()
{
    m_vecSharedBuffer.resize(32 * 4 * 1024);//128k共享缓存
    //给本例创建对应的epoll
    m_hEpoll = epoll_create(EPOLL_SIZE);
}

EventPoller::~EventPoller()
{
}

bool EventPoller::AddEvent(int fd, int event, PollEventCB cb)
{
    //如果是当前线程进行的事件添加则直接放进map里
    struct epoll_event ev = { 0 };
    ev.events = event;
    ev.data.fd = fd;
    const int iRet = epoll_ctl(m_hEpoll, EPOLL_CTL_ADD, fd, &ev);
    if (-1 == iRet) {
        return false;
    }
    //记录要监听的套接字和其对应的处理回调
    m_mapEventCB.emplace(fd, std::make_shared<PollEventCB>(std::move(cb)));
    return true;
}

bool EventPoller::DelEvent(int fd) {

    int iRet = -1;
    //仅删除监听中的fd，如果没删除成功则意味着对应的fd在逻辑上已不受监听而无需DEL。
    //如果后续发现在监听回调里出现了不在监听中的fd，则在监听回调里再进行DEL，同时忽略那次的监听执行。
    if (m_mapEventCB.erase(fd)) {
        iRet = epoll_ctl(m_hEpoll, EPOLL_CTL_DEL, fd, nullptr);
    }
    return -1 != iRet;
}

void EventPoller::Exec()
{
    m_isRun = true;
    int iMinDelay = 5000;//等待事件的超时毫秒数
    struct epoll_event *events = new struct epoll_event[EPOLL_SIZE];
    while (m_isRun) {
        int ret = epoll_wait(m_hEpoll, events, EPOLL_SIZE, iMinDelay);
        if (ret <= 0) 
            continue;

        for (int i = 0; i < ret; ++i) {
            struct epoll_event& ev = events[i];
            int fd = ev.data.fd;

            //找找要被监听的事件map里有没有这个fd，没有就移除掉这个残留的监听
            auto it = m_mapEventCB.find(fd);
            if (it == m_mapEventCB.end()) {
                epoll_ctl(m_hEpoll, EPOLL_CTL_DEL, fd, nullptr);
                continue;
            }
            (*(it->second))(ev.events);
        }
    }
    delete[] events;
}

std::vector<char>& EventPoller::GetSharedBuffer()
{
    return m_vecSharedBuffer;
}

4.RtspServer.h

#pragma once
#include 
#include 
#include 

class EventPoller;
class RtspSession;

class RtspServer :  public std::enable_shared_from_this<RtspServer> {
public:
    explicit RtspServer(EventPoller* pEventPoller);
    /// 
    /// 启动服务器
    /// 
    /// 主机
    /// 端口
    /// 
    bool Start(const std::string& szHost, uint16_t uPort);
    //停止服务器运行
    void Stop();
    //关闭指定的session连接
    void CloseSession(int fd);

private:
    EventPoller* m_pEventPoller = nullptr;    //监听事件的实例
    std::string m_szHost;   //服务器监听ip
    uint16_t m_uPort = 0;   //服务器端口
    int m_fd = -1;          //服务器套接字
    std::unordered_map<int, std::shared_ptr<RtspSession>> m_mapSession; //保存所有的连接session
    std::mutex m_mtxSession; //保护session容器的互斥量
};

5.RtspServer.cpp

#pragma once
#include "RtspServer.h"

#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include 
#include 
#include 
#pragma comment (lib, "Ws2_32.lib")
#pragma comment(lib,"Iphlpapi.lib")
#include 
#include 
#include 
#include 

#include "EventPoller.h"
#include "RtspSession.h"

using namespace std;

RtspServer::RtspServer(EventPoller* pEventPoller)
{
	m_pEventPoller = pEventPoller;
}

bool RtspServer::Start(const std::string& szHost, uint16_t uPort)
{
    m_szHost = szHost;
    m_uPort = uPort;

    m_fd = (int)socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

    int opt = 1;
    //设置套接字属性，此处处理服务器快速重启后的以前使用的套接字还在TIME_WAIT的问题，配置后可以直接接着使用
    setsockopt(m_fd, SOL_SOCKET, SO_REUSEADDR, (char*)&opt, static_cast<socklen_t>(sizeof(opt)));
    unsigned long ul = 1; //设置为非阻塞模式
    ioctlsocket(m_fd, FIONBIO, &ul);

    struct sockaddr_in addr;
    ZeroMemory(&addr, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(uPort);
    addr.sin_addr.s_addr = INADDR_ANY;
    if (::bind(m_fd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
        DebugBreak();
    }

    if (::listen(m_fd, 1024) == -1) {
        DebugBreak();
    }
    //weak_self的存在是为了防止发生回调时，本类的实例已经析构的情况
    //这是因为m_pEventPoller->DelEvent只是停止监听新事件，但队列里的事件仍然有可能在实例析构后仍触发该回调
    //此时必须忽略该回调，否则会访问野指针
    std::weak_ptr<RtspServer> weak_self = shared_from_this();
    //这个事件处理主要用于accpet新的套接字
    bool isRet = m_pEventPoller->AddEvent(m_fd, EventPoller::Event_Read | EventPoller::Event_Error, [weak_self](int event) {
        auto strong_self = weak_self.lock();
        if (nullptr == strong_self)
            return;
        const int fdServer = strong_self->m_fd;
        int fd;
        struct sockaddr_storage peer_addr;
        socklen_t addr_len = sizeof(peer_addr);
        //这个while主要是用来尝试一次获取多个accpet新连接
        while (true) {
            if (event & EventPoller::Event_Read) {
                fd = (int)accept(fdServer, (struct sockaddr*)&peer_addr, &addr_len);
                if (fd == -1) {// accept失败
                    int iRet = WSAGetLastError();
                    if (iRet == WSAEWOULDBLOCK) {
                        //没有新的连接accept进来了
                        return;
                    }
                    DebugBreak();
                    return;
                }
                auto session = std::make_shared<RtspSession>(strong_self, strong_self->m_pEventPoller, fd);
                if (session->StartSession()) {
                    strong_self->m_mtxSession.lock();
                    strong_self->m_mapSession[fd] = session;
                    strong_self->m_mtxSession.unlock();
                }
            }
            if (event & EventPoller::Event_Error) {
                return;
            }
        }
        });
    //没附加事件处理成功就直接停止服务器运行
    if (false == isRet)
        Stop();
    return isRet;
}

void RtspServer::Stop()
{
    if (-1 != m_fd) {
        m_pEventPoller->DelEvent(m_fd);
        closesocket(m_fd);
        m_fd = -1;
    }
}

void RtspServer::CloseSession(int fd)
{
    thread task([this,fd]() {
        m_mtxSession.lock();
        m_mapSession.erase(fd);
        m_mtxSession.unlock();
        });
    task.detach();
}

6.RtspSession.h

#pragma once

#include 
#include 
#include 
#include 
#include "SdpParser.h"

#pragma once

class EventPoller;
class RtspServer;

/// 
/// 对于发往RtspServer的每一个连接，都会对应一个RtspSession实例，用于保存该连接的状态信息
/// 
class RtspSession : public std::enable_shared_from_this<RtspSession> {
public:
    RtspSession(std::weak_ptr<RtspServer> pServer, EventPoller* pEventPoller, int fd);
    virtual ~RtspSession();
    //附加事件监听从而启动该session的处理
    bool StartSession();
    //向服务器通知关闭该session
    void CloseSession();
    //发送数据给该session对应的fd
    bool SendData(const char* buffer, size_t uSize);
protected:
    //处理给定的套接字的读取请求，尽可能地获取数据后交给onRecv进一步处理
    size_t onRead(std::vector<char>& buffer);
    //接收并处理fd对应的数据
    void onRecv(char* buffer, size_t uSize);
private:
    //解析请求头
    bool parseHeader();
    //处理一次完整的请求
    void handleReq();
    //处理rtp包
    void handleRtpPacket();
private:
    // 处理options方法,获取服务器能力
    void handleReq_Options(const std::unordered_map<std::string, std::string>& parser);
    // 处理describe方法，请求服务器rtsp sdp信息 
    void handleReq_Describe(const std::unordered_map<std::string, std::string>& parser);
    // 处理ANNOUNCE方法，请求推流，附带sdp 
    void handleReq_ANNOUNCE(const std::unordered_map<std::string, std::string>& parser);
    // 处理record方法，开始推流 
    void handleReq_RECORD(const std::unordered_map<std::string, std::string>& parser);
    // 处理setup方法，播放和推流协商rtp传输方式用 
    void handleReq_SETUP(const std::unordered_map<std::string, std::string>& parser);
    // 处理play方法，开始或恢复播放 
    void handleReq_PLAY(const std::unordered_map<std::string, std::string>& parser);
    // 处理teardown方法，结束播放 
    void handleReq_TEARDOWN(const std::unordered_map<std::string, std::string>& parser);
    // 处理SET_PARAMETER、GET_PARAMETER方法，一般用于心跳保活
    void handleReq_SET_PARAMETER(const std::unordered_map<std::string, std::string>& parser);


    // 回复客户端
    bool sendRtspResponse(const std::string& res_code, std::multimap<std::string, std::string> header = {}, const std::string& sdp = "", const char* protocol = "RTSP/1.0");

    // 获取track下标 
    size_t getTrackIndexByControlUrl(const std::string& control_url);

private:
    std::weak_ptr<RtspServer> m_pServer;         //session所属的服务器
    EventPoller* m_pEventPoller = nullptr;      //监听事件的实例
    int m_fd = -1;	                            //连接所对应的套接字
    std::vector<char> m_vecRecv;    //接收缓冲
    std::vector<char> m_vecHeader;  //接收结果:请求头
    std::vector<char> m_vecContent; //接收结果:请求内容
    bool m_isHeader = true;         //当前是否仍在接收请求头
    int m_iContentLength = 0;		//附带的内容长度
    size_t m_uRtpPacketLength = 0;	//正在接收的rtp包长度
    std::unordered_map<std::string, std::string> m_mapHeader;	//请求头的map
    // 收到的seq，回复时一致
    int m_iCseq = 0;
    // Session号
    std::string m_szSessionId;
    // 解析sdp的实例
    SdpParser* m_pSdpParser = nullptr;
    // sdp里面有效的track,包含音频或视频(每样最多一个)
    std::vector<SdpTrack*> m_vecSdpTrack;
    //完整链接（含后面参数）
    std::string m_szFullUrl;
    //该链接是否是推流方
    bool m_isPusher = false;
    Rtsp::ERtpType m_eRtpType = Rtsp::ERtpType::INVALID;

    //////////////////拉流时的session使用变量
    RtspSession* m_pPusherRtspSession = nullptr;    //在拉取谁的推流
};

7.RtspSession.cpp

#include "RtspSession.h"

#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include 
#include 
#pragma comment (lib, "Ws2_32.lib")
#include 
#include 
#include 
#include 
#include 
#include 
#include 

#include "EventPoller.h"
#include "MediaPusherManager.h"
#include "RtspServer.h"

using namespace std;

std::string makeRandStr(size_t uLen)
{
	static constexpr char CCH[] = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
	static constexpr size_t CCH_LEN = sizeof(CCH) - 1; // 预计算长度，避免重复计算
	string ret;
	ret.resize(uLen);
	thread_local std::mt19937 rng(std::random_device{}());
	for (size_t i = 0; i < uLen; ++i) {
		ret[i] = CCH[rng() % (sizeof(CCH) - 1)];
	}
	return ret;
}

//获取当前时间的字符串
static string dateStr() {
	char buf[64];
	time_t tt = time(NULL);
	tm t;
	gmtime_s(&t, &tt);
	strftime(buf, sizeof buf, "%a, %b %d %Y %H:%M:%S GMT", &t);
	return buf;
}

RtspSession::RtspSession(std::weak_ptr<RtspServer> pServer, EventPoller* pEventPoller, int fd)
	:m_pServer(pServer)
	, m_pEventPoller(pEventPoller)
	, m_fd(fd)
{
	unsigned long ul = 1; //设置为非阻塞模式
	ioctlsocket(fd, FIONBIO, &ul);
	int opt = 1;
	setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char*)&opt, static_cast<socklen_t>(sizeof(opt)));
	int size = 262144;
	setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&size, sizeof(size));
	setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char*)&size, sizeof(size));
	linger m_sLinger;
	//在调用closesocket()时还有数据未发送完，允许等待
	// 若m_sLinger.l_onoff=0;则调用closesocket()后强制关闭 
	m_sLinger.l_onoff = false;
	m_sLinger.l_linger = 0; //设置等待时间为x秒
	setsockopt(fd, SOL_SOCKET, SO_LINGER, (char*)&m_sLinger, sizeof(linger));
}

RtspSession::~RtspSession()
{
	if (m_isPusher)MediaPusherManager::Instance().DelPusher(m_szFullUrl);
	else MediaPusherManager::Instance().DelRecvier(m_szFullUrl, this);
}

bool RtspSession::StartSession()
{
	//给客户端套接字接入监听
	std::weak_ptr<RtspSession> weak_self = shared_from_this();
	bool isRet = m_pEventPoller->AddEvent(m_fd, EventPoller::Event_Read | EventPoller::Event_Error
		, [weak_self](int event) {
			auto strong_self = weak_self.lock();
			if (!strong_self) {
				//来到这里，意味着客户端本身已经析构，这次回调只是队列残留，忽略即可
				return;
			}
			if (event & EventPoller::Event_Read) {
				strong_self->onRead(strong_self->m_pEventPoller->GetSharedBuffer());
			}
			if (event & EventPoller::Event_Error) {
				strong_self->CloseSession();
			}
		});
	return isRet;
}

void RtspSession::CloseSession()
{
	m_pEventPoller->DelEvent(m_fd);
	auto pServer = m_pServer.lock();
	if (nullptr == pServer)
		return;
	//通知服务器把自己关了
	pServer->CloseSession(m_fd);
}

bool RtspSession::SendData(const char* buffer, size_t uSize)
{
	const int iSendRet = ::send(m_fd, buffer, uSize, 0);
	return iSendRet > 0;
}

size_t RtspSession::onRead(std::vector<char>& buffer)
{
	size_t uReadSizeAll = 0, uReadSize = 0, count = 0;
	while (true) {
		uReadSize = (size_t)recv(m_fd, buffer.data(), buffer.size() - 1, 0);
		if (uReadSize == 0) {
			//来到这里意味着套接字已被对方关闭，正常关闭回收就行
			CloseSession();
			return uReadSizeAll;
		}
		if (uReadSize == -1) {
			int iRetWSA = WSAGetLastError();
			if (iRetWSA != WSAEWOULDBLOCK) {
				CloseSession();
			}
			return uReadSizeAll;
		}
		buffer[uReadSize] = '';
		uReadSizeAll += uReadSize;
		//交给子类实现具体处理这些数据
		onRecv(buffer.data(), uReadSize);
	}
}

void RtspSession::onRecv(char* buffer, size_t uSize)
{
	size_t uIndex = 0;


	//进入后续处理前，至少有4个缓存字节用于判断
	if (m_vecRecv.size() < 4) {
		size_t uLen = min(4 - m_vecRecv.size(), uSize);
		m_vecRecv.insert(m_vecRecv.end(), buffer, buffer + uLen);
		if (m_vecRecv.size() < 4)
			return;//给定的数据不足，则记录已有的后就直接返回
		uSize -= uLen;
		buffer += uLen;
	}

	//检查是否是要处理rtp包
	if (m_vecRecv[0] == '$') {
		char* data = m_vecRecv.data();
		size_t uDataLen = m_vecRecv.size();
		//确定rtp包的接收大小
		if (0 == m_uRtpPacketLength) {
			m_uRtpPacketLength = ((((uint8_t*)data)[2] << 8) | ((uint8_t*)data)[3]) + 4;
		}

		size_t uLen = min(m_uRtpPacketLength - uDataLen, uSize);
		m_vecRecv.insert(m_vecRecv.end(), buffer, buffer + uLen);
		uSize -= uLen;
		buffer += uLen;

		//是否已经足够一个rtp包可以开始处理了
		if (m_vecRecv.size() == m_uRtpPacketLength) {
			handleRtpPacket();
			if (uSize > 0)//把剩下的数据也处理掉
				onRecv(buffer, uSize);
		}
		return;
	}

	//处理请求，而非rtp包
	if (m_isHeader) {
		while (uIndex < uSize) {
			m_vecRecv.push_back(buffer[uIndex]);
			if (buffer[uIndex] == '
') {
				//检查是否以rnrn结尾
				const size_t vec_size = m_vecRecv.size();
				if (m_vecRecv[vec_size - 1] == '
'
					&& m_vecRecv[vec_size - 2] == '
'
					&& m_vecRecv[vec_size - 3] == '
'
					&& m_vecRecv[vec_size - 4] == '
') {
					m_vecRecv.push_back('');
					m_vecHeader.swap(m_vecRecv);
					m_vecRecv.clear();
					++uIndex;
					//解析请求头
					if (parseHeader()) {
						//如果有附带内容，则继续解析
						if (m_iContentLength > 0) {
							m_isHeader = false;
						}
						else {//否则直接使用该请求头进行请求处理
							handleReq();
						}
					}
					else {
						m_vecHeader.clear();
					}
					break;//请求头获取结束
				}
			}
			++uIndex;
		}
	}
	if (false == m_isHeader) {
		while (uIndex < uSize) {
			m_vecRecv.push_back(buffer[uIndex]);
			//如果已经接收到了足够的内容
			if (m_vecRecv.size() == m_iContentLength) {
				m_vecRecv.push_back('');
				m_vecContent.swap(m_vecRecv);
				m_vecRecv.clear();
				m_mapHeader["reqContent"] = m_vecContent.data();
				m_isHeader = true;
				++uIndex;
				handleReq();
				break;//请求体获取结束
			}
			++uIndex;
		}
	}
	//如果发现还有数据则进行递归处理
	if (uSize - uIndex != 0)
		onRecv(buffer + uIndex, uSize - uIndex);
}

bool RtspSession::parseHeader()
{
	/*
	OPTIONS rtsp://192.168.114.114:554/live/test RTSP/1.0
	CSeq: 1
	User-Agent: Lavf58.76.100
	*/
	m_mapHeader.clear();
	std::istringstream stream(m_vecHeader.data());
	stream >> m_mapHeader["reqMethod"];
	stream >> m_mapHeader["reqUrl"];
	stream >> m_mapHeader["reqProtocol"];
	std::string szUrl = m_mapHeader.at("reqUrl");
	size_t uSplitIndex = szUrl.find('?');
	if (std::string::npos != uSplitIndex) {
		m_mapHeader["reqUrl_Short"] = szUrl.substr(0, uSplitIndex);
		m_mapHeader["reqUrl_Args"] = szUrl.substr(uSplitIndex + 1);
	}
	else {
		m_mapHeader["reqUrl_Short"] = szUrl;
	}
#define RM_R(STR) STR = STR.substr(0, STR.size() - 1)
	std::string line;
	while (std::getline(stream, line)) {
		size_t pos = line.find(':');
		if (pos != std::string::npos) {
			std::string key = line.substr(0, pos);
			std::string value = line.substr(pos + 1);
			// 去掉首尾空格
			key.erase(0, key.find_first_not_of(" 	"));
			key.erase(key.find_last_not_of(" 	") + 1);
			value.erase(0, value.find_first_not_of(" 	"));
			value.erase(value.find_last_not_of(" 	") + 1);
			RM_R(value);
			m_mapHeader[key] = value;
		}
	}
#undef RM_R
	if (0 == m_mapHeader.size()) {
		return false;
	}
	m_iContentLength = atoi(m_mapHeader["Content-Length"].c_str());
	return true;
}

void RtspSession::handleReq()
{
	auto method = m_mapHeader["reqMethod"];
	m_iCseq = atoi(m_mapHeader["CSeq"].data());
	if (m_szFullUrl.empty()) {
		m_szFullUrl = m_mapHeader.at("reqUrl");
	}
	using rtsp_request_handler = void (RtspSession::*)(const std::unordered_map<std::string, std::string>& parser);
	static unordered_map<string, rtsp_request_handler> s_cmd_functions{
		{"OPTIONS", &RtspSession::handleReq_Options},//拉流推流都会用到
		{"DESCRIBE", &RtspSession::handleReq_Describe},//拉流会用到
		{"ANNOUNCE", &RtspSession::handleReq_ANNOUNCE},//推流会用到
		{"RECORD", &RtspSession::handleReq_RECORD},//推流会用到
		{"SETUP", &RtspSession::handleReq_SETUP},//拉流推流都会用到
		{"PLAY", &RtspSession::handleReq_PLAY},
		{"TEARDOWN", &RtspSession::handleReq_TEARDOWN},//拉流推流都会用到
		{"GET_PARAMETER", &RtspSession::handleReq_SET_PARAMETER},//保活
	};
	auto it = s_cmd_functions.find(method);
	if (it == s_cmd_functions.end()) {
		sendRtspResponse("403 Forbidden");
	}
	else {
		(this->*(it->second))(m_mapHeader);
	}
	m_vecHeader.clear();
	m_vecContent.clear();
	m_iContentLength = 0;
}

void RtspSession::handleRtpPacket()
{
	////("{}字节rtp包到达", m_vecRecv.size());
	char* data = m_vecRecv.data();
	size_t uDataLen = m_vecRecv.size();

	//直接转发
	MediaPusherManager::Instance().SendDataToRecvier(m_szFullUrl, data, uDataLen);

	m_vecRecv.clear();
	m_uRtpPacketLength = 0;
}

void RtspSession::handleReq_Options(const std::unordered_map<std::string, std::string>& parser) {
	/*
	21444][2025-12-12 10:38:06.986][13696][RtspSession.cpp:21, onRecv][I]   OPTIONS rtsp://192.168.114.114:554/live/test RTSP/1.0
	CSeq: 1
	User-Agent: Lavf58.76.100
	*/
	//支持这些命令
	std::multimap<std::string, std::string> header;
	header.emplace("Public", "OPTIONS, DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE, ANNOUNCE, RECORD, SET_PARAMETER, GET_PARAMETER");
	sendRtspResponse("200 OK", header);
}

void RtspSession::handleReq_Describe(const std::unordered_map<std::string, std::string>& parser) {
	m_pPusherRtspSession = dynamic_cast<RtspSession*>(MediaPusherManager::Instance().AddRecvier(m_szFullUrl, this));
	if (nullptr == m_pPusherRtspSession) {
		static constexpr auto err = "该流在服务器上不存在";
		std::multimap<std::string, std::string> header;
		header.emplace("Content-Type", "text/plain");
		sendRtspResponse("406 Not Acceptable", header, err);
		//("{}:{}", err, m_szFullUrl);
		return;
	}
	m_isPusher = false;
	//("观看流拉流成功:{}", m_szFullUrl);
	//找到了相应的rtsp流
	m_pSdpParser = new SdpParser(m_pPusherRtspSession->m_pSdpParser->GetSdp());
	m_vecSdpTrack = m_pSdpParser->GetAvailableTrack();
	if (m_vecSdpTrack.empty()) {
		//该流无效
		static constexpr auto err = "sdp中无有效track，该流无效";
		std::multimap<std::string, std::string> header;
		header.emplace("Content-Type", "text/plain");
		sendRtspResponse("406 Not Acceptable", header, err);
		//("{}:{}", err, m_szFullUrl);
		return;
	}
	m_szSessionId = makeRandStr(12);

	std::multimap<std::string, std::string> header;
	header.emplace("Content-Base", m_szFullUrl + "/");
	header.emplace("x-Accept-Retransmit", "our-retransmit");
	header.emplace("x-Accept-Dynamic-Rate", "1");
	sendRtspResponse("200 OK", header, m_pSdpParser->GetSdp());
}

void RtspSession::handleReq_ANNOUNCE(const std::unordered_map<std::string, std::string>& parser)
{
	/*
	[21444][2025-12-12 10:38:10.129][13696][RtspSession.cpp:21, onRecv][I]   ANNOUNCE rtsp://192.168.114.114:554/live/test RTSP/1.0
	Content-Type: application/sdp
	CSeq: 2
	User-Agent: Lavf58.76.100
	Content-Length: 296

	v=0
	o=- 0 0 IN IP4 127.0.0.1
	s=No Name
	c=IN IP4 192.168.114.114
	t=0 0
	a=tool:libavformat 58.76.100
	m=video 0 RTP/AVP 96
	a=rtpmap:96 H264/90000
	a=fmtp:96 packetization-mode=1; sprop-parameter-sets=Z/QAMpGbKAeAET8TCAAAH0gAB1MAeMGMsA==,aOvjxEhE; profile-level-id=F40032
	a=control:streamid=0
	*/

	m_szFullUrl = parser.at("reqUrl");
	if (false == MediaPusherManager::Instance().AddPusher(m_szFullUrl, this)) {
		static constexpr auto err = "该流在服务器上已存在";
		std::multimap<std::string, std::string> header;
		header.emplace("Content-Type", "text/plain");
		sendRtspResponse("406 Not Acceptable", header, err);
		//("{}:{}", err, m_szFullUrl);
		return;
	}
	m_isPusher = true;
	//("流添加成功:{}", m_szFullUrl);

	m_pSdpParser = new SdpParser(parser.at("reqContent"));
	m_szSessionId = makeRandStr(12);
	m_vecSdpTrack = m_pSdpParser->GetAvailableTrack();
	if (m_vecSdpTrack.empty()) {
		static constexpr auto err = "sdp中无有效track";
		std::multimap<std::string, std::string> header;
		header.emplace("Content-Type", "text/plain");
		sendRtspResponse("403 Forbidden", header, err);
		//("{}:{}", err, m_szFullUrl);
		return;
	}
	sendRtspResponse("200 OK");
}

void RtspSession::handleReq_RECORD(const std::unordered_map<std::string, std::string>& parser)
{
	if (m_vecSdpTrack.empty() || parser.at("Session") != m_szSessionId) {
		std::multimap<std::string, std::string> header;
		header.emplace("Connection", "Close");
		sendRtspResponse("454 Session Not Found", header);
		//("{}:{}", "454 Session Not Found", m_szFullUrl);
		return;
	}

	stringstream rtp_info;
	for (auto& track : m_vecSdpTrack) {
		if (track->isInited == false) {
			//还有track没有setup
			static constexpr auto err = "track not setuped";
			std::multimap<std::string, std::string> header;
			header.emplace("Content-Type", "text/plain");
			sendRtspResponse("403 Forbidden", header, err);
			//("{}:{}", err, m_szFullUrl);
			return;
		}
		rtp_info << "url=" << track->GetControlUrl(m_szFullUrl) << ",";
	}
	auto rtpStr = rtp_info.str();
	rtpStr.pop_back();
	std::multimap<std::string, std::string> header;
	header.emplace("RTP-Info", rtpStr);
	sendRtspResponse("200 OK", header);
}

void RtspSession::handleReq_SETUP(const std::unordered_map<std::string, std::string>& parser)
{
	/*
	SETUP rtsp://192.168.114.114:554/live/test/streamid=0 RTSP/1.0
	Transport: RTP/AVP/TCP;unicast;interleaved=0-1;mode=record
	CSeq: 3
	User-Agent: Lavf58.76.100
	Session: uMPQjlcMcI7J
	*/
	//处理setup命令，该函数可能进入多次m_szFullUrl

	int trackIdx = getTrackIndexByControlUrl(parser.at("reqUrl"));
	SdpTrack* trackRef = m_vecSdpTrack[trackIdx];
	if (trackRef->isInited) {
		//已经初始化过该Track
		static constexpr auto err = "不允许对同一个track进行两次setup";
		std::multimap<std::string, std::string> header;
		header.emplace("Content-Type", "text/plain");
		sendRtspResponse("403 Forbidden", header, err);
		//("{}:{}", err, m_mediaTuple.ShortUrl());
		return;
	}

	static auto getRtpTypeStr = [](const Rtsp::ERtpType type) {
		switch (type)
		{
		case Rtsp::ERtpType::TCP:
			return "TCP";
		case Rtsp::ERtpType::UDP:
			return "UDP";
		case Rtsp::ERtpType::MULTICAST:
			return "MULTICAST";
		default:
			return "Invalid";
		}
		};

	if (m_eRtpType == Rtsp::ERtpType::INVALID) {
		auto& strTransport = parser.at("Transport");
		auto rtpType = Rtsp::ERtpType::INVALID;
		if (strTransport.find("TCP") != string::npos) {
			rtpType = Rtsp::ERtpType::TCP;
		}
		else if (strTransport.find("multicast") != string::npos) {
			//rtpType = Rtsp::ERtpType::MULTICAST;
		}
		else {
			//rtpType = Rtsp::ERtpType::UDP;
		}
		if (Rtsp::ERtpType::INVALID == rtpType) {
			sendRtspResponse("461 Unsupported transport");
			return;
		}
		m_eRtpType = rtpType;
	}

	trackRef->isInited = true; //现在初始化

	{
		// rtsp推流时，interleaved由推流者决定
		auto key_values = SdpParser::ParseArgs(parser.at("Transport"), ";", "=");
		int interleaved_rtp = -1, interleaved_rtcp = -1;
		if (2 == sscanf_s(key_values["interleaved"].data(), "%d-%d", &interleaved_rtp, &interleaved_rtcp)) {
			trackRef->uInterleaved = interleaved_rtp;
		}
		else {
			static constexpr auto err = "can not find interleaved when setup of rtp over tcp";
			std::multimap<std::string, std::string> header;
			header.emplace("Content-Type", "text/plain");
			sendRtspResponse("403 Forbidden", header, err);
			//("{}:{}", err, m_mediaTuple.ShortUrl());
			return;
		}

		stringstream sdpResponse;
		sdpResponse << "RTP/AVP/TCP;unicast;"
			<< "interleaved=" << (int)trackRef->uInterleaved << "-"
			<< (int)trackRef->uInterleaved + 1 << ";"
			<< "ssrc=00000000";

		std::multimap<std::string, std::string> header;
		header.emplace("Transport", sdpResponse.str());
		header.emplace("x-Transport-Options", "late-tolerance=1.400000");
		header.emplace("x-Dynamic-Rate", "1");

		sendRtspResponse("200 OK", header);
	}
}

void RtspSession::handleReq_PLAY(const std::unordered_map<std::string, std::string>& parser) {
	if (m_vecSdpTrack.empty() || parser.at("Session") != m_szSessionId) {
		std::multimap<std::string, std::string> header;
		header.emplace("Connection", "Close");
		sendRtspResponse("454 Session Not Found", header);
		//("{}:{}", "454 Session Not Found", m_szFullUrl);
		return;
	}

	std::multimap<std::string, std::string> res_header;

	vector<ETrackType> inited_tracks;
	stringstream rtp_info;
	for (auto& track : m_vecSdpTrack) {
		if (track->isInited == false) {
			//为支持播放器播放单一track, 不校验没有发setup的track
			continue;
		}
		inited_tracks.emplace_back(track->type);

		rtp_info << "url=" << track->GetControlUrl(m_szFullUrl) << ";"
			<< "seq=0;"
			<< "rtptime=0" << ",";
	}

	auto rtpStr = rtp_info.str();
	rtpStr.pop_back();

	res_header.emplace("RTP-Info", rtpStr);
	//已存在Range时不覆盖
	stringstream szPlayRange;
	szPlayRange << "npt=" << setiosflags(ios::fixed) << setprecision(2) << 0;
	res_header.emplace("Range", szPlayRange.str());
	sendRtspResponse("200 OK", res_header);
}

void RtspSession::handleReq_TEARDOWN(const std::unordered_map<std::string, std::string>& parser)
{
	sendRtspResponse("200 OK");
	CloseSession();
}

void RtspSession::handleReq_SET_PARAMETER(const std::unordered_map<std::string, std::string>& parser)
{
	sendRtspResponse("200 OK");
}

bool RtspSession::sendRtspResponse(const std::string& res_code, std::multimap<std::string, std::string> header, const std::string& sdp, const char* protocol)
{
	header.emplace("CSeq", to_string(m_iCseq));
	if (!m_szSessionId.empty()) {
		header.emplace("Session", m_szSessionId);
	}

	header.emplace("Server", "服务器名字");
	header.emplace("Date", dateStr());

	if (!sdp.empty()) {
		header.emplace("Content-Length", to_string(sdp.size()));
		header.emplace("Content-Type", "application/sdp");
	}

	stringstream printer;
	printer << protocol << " " << res_code << "
";
	for (auto& pr : header) {
		printer << pr.first << ": " << pr.second << "
";
	}

	printer << "
";

	if (!sdp.empty()) {
		printer << sdp;
	}
	const int iSendRet = ::send(m_fd, printer.str().data(), printer.str().size(), 0);
	return iSendRet > 0;
}

size_t RtspSession::getTrackIndexByControlUrl(const std::string& control_url)
{
	for (size_t i = 0; i < m_vecSdpTrack.size(); ++i) {
		if (control_url.find(m_vecSdpTrack[i]->GetControlUrl(m_szFullUrl)) == 0) {
			return i;
		}
	}
	if (m_vecSdpTrack.size() == 1) {
		return 0;
	}
	DebugBreak();
}

8.SdpParser.h

#pragma once

#include 
#include 
#include 

//通道类型
enum class ETrackType {
    INVALID = -1,  //无效通道
    VIDEO = 0,     //视频通道
    AUDIO,         //音频通道
    TITLE,         //原始通道
    MIN = INVALID,
    MAX = TITLE,
};

namespace Rtsp {
    //RTSP的RTP类型
    enum class ERtpType {
        INVALID = -1,
        TCP = 0,
        UDP = 1,
        MULTICAST = 2,
    };
};

//一个通道的具体属性
struct SdpTrack {
    std::multimap<std::string, std::string> map_attr;  //通道属性
    ETrackType type = ETrackType::INVALID; //这个track的类型
    std::string szControl;//媒体流的控制属性。当存在属性control时可用
    uint8_t uInterleaved = 0;   //rtp的通道号
    bool isInited = false;   //该通道是否已经被setup过

    std::string GetControlUrl(const std::string& szUrl) const;
};

class SdpParser {
public:
    SdpParser(const std::string& szSdp);
    ~SdpParser();
    //获取第一个指定类型的通道
    SdpTrack* GetTrack(ETrackType type) const;
    //取第一个有效的视频通道和音频通道，最多两个元素
    std::vector<SdpTrack*> GetAvailableTrack() const;
    //获取sdp的解析原文
    const std::string GetSdp() const { return m_szSdp; }
    //解析参数为map
    static std::map<std::string, std::string> ParseArgs(const std::string& str, const char* pair_delim, const char* key_delim);

private:
    std::string m_szSdp;
    std::vector<SdpTrack*> m_vecTrack;
};

9.SdpParser.cpp

#include "SdpParser.h"
#include 

using namespace std;

static int GetClockRate(int pt)
{
    switch (pt) {
    case 0:  return 8000;   // PCMU
    case 3:  return 8000;   // GSM
    case 4:  return 8000;   // G723
    case 5:  return 8000;   // DVI4_8000
    case 6:  return 16000;  // DVI4_16000
    case 7:  return 8000;   // LPC
    case 8:  return 8000;   // PCMA
    case 9:  return 16000;  // G722
    case 10: return 44100;  // L16_Stereo
    case 11: return 44100;  // L16_Mono
    case 12: return 8000;   // QCELP
    case 13: return 8000;   // CN
    case 14: return 44100;  // MP3
    case 15: return 8000;   // G728
    case 16: return 11025;  // DVI4_11025
    case 17: return 22050;  // DVI4_22050
    case 18: return 8000;   // G729
    case 25: return 90000;  // CelB (视频)
    case 26: return 90000;  // JPEG (视频)
    case 28: return 90000;  // nv (视频)
    case 31: return 90000;  // H261 (视频)
    case 32: return 90000;  // MPV (视频)
    case 33: return 90000;  // MP2T (视频)
    case 34: return 90000;  // H263 (视频)
    default: return 90000;  // 默认返回90000（视频默认采样率/时钟频率）
    }
}

static vector<string> split(const string& s, const char* delim) {
    vector<string> ret;
    size_t last = 0;
    auto index = s.find(delim, last);
    while (index != string::npos) {
        if (index - last > 0) {
            ret.push_back(s.substr(last, index - last));
        }
        last = index + strlen(delim);
        index = s.find(delim, last);
    }
    if (!s.size() || s.size() - last > 0) {
        ret.push_back(s.substr(last));
    }
    return ret;
}

//去除前后的空格、回车符、制表符
static string& trim(string& s, const string& chars = " 
	") {
    string map(0xFF, '');
    for (auto& ch : chars) {
        map[(unsigned char&)ch] = '';
    }
    while (s.size() && map.at((unsigned char&)s.back())) s.pop_back();
    while (s.size() && map.at((unsigned char&)s.front())) s.erase(0, 1);
    return s;
}

static string findSubString(const char* buf, const char* start, const char* end, size_t buf_size = 0) {
    if (buf_size <= 0) {
        buf_size = strlen(buf);
    }
    auto msg_start = buf;
    auto msg_end = buf + buf_size;
    size_t len = 0;
    if (start != NULL) {
        len = strlen(start);
        msg_start = strstr(buf, start);
    }
    if (msg_start == NULL) {
        return "";
    }
    msg_start += len;
    if (end != NULL) {
        msg_end = strstr(msg_start, end);
        if (msg_end == NULL) {
            return "";
        }
    }
    return string(msg_start, msg_end);
}

static ETrackType toTrackType(const string& str) {
    if (str == "") {
        return ETrackType::TITLE;
    }

    if (str == "video") {
        return ETrackType::VIDEO;
    }

    if (str == "audio") {
        return ETrackType::AUDIO;
    }

    return ETrackType::INVALID;
}

SdpParser::SdpParser(const std::string& szSdp)
    :m_szSdp(szSdp)
{
    /*****
    v=0 #SDP 的版本号，目前固定为 0，是唯一的有效值。
    o=- 0 0 IN IP4 127.0.0.1 #o=<用户名:匿名> <会话ID:默认> <版本号:默认> <网络类型:互联网（IN）> <地址类型: IPv4> <地址:本地回环>
    s=No Name #	会话的名称
    c=IN IP4 192.168.114.114 #媒体流的传输地址
    t=0 0 #表示会话永久有效
    a=tool:libavformat 58.76.100 #生成这份 SDP 的工具是libavformat（FFmpeg 的核心格式处理库），版本号为 58.76.100
    m=video 0 RTP/AVP 96 #m=<媒体类型> <端口> <传输协议> <编码载荷类型:1-95 为标准载荷，96 + 为自定义>
    a=rtpmap:96 H264/90000 #关联载荷类型和具体编码格式
    a=fmtp:96 packetization-mode=1; sprop-parameter-sets=Z/QAMpGbKAeAET8TCAAAH0gAB1MAeMGMsA==,aOvjxEhE; profile-level-id=F40032
    a=control:streamid=0
    *****/

    SdpTrack* track = new SdpTrack;
    track->type = ETrackType::TITLE;
    m_vecTrack.emplace_back(track);

    auto lines = split(m_szSdp, "
");
    for (auto& line : lines) {
        trim(line);
        if (line.size() < 2 || line[1] != '=') {
            continue;
        }
        char opt = line[0];
        string opt_val = line.substr(2);
        switch (opt) {
        case 'm': {
            track = new SdpTrack;
            m_vecTrack.emplace_back(track);
            int pt, port, port_count;
            char rtp[16] = { 0 }, type[16] = { 0 };
            if (4 == sscanf_s(opt_val.data(), " %15[^ ] %d %15[^ ] %d", type, (unsigned)_countof(type), &port, rtp, (unsigned)_countof(rtp), &pt) ||
                5 == sscanf_s(opt_val.data(), " %15[^ ] %d/%d %15[^ ] %d", type, (unsigned)_countof(type), &port, &port_count, rtp, (unsigned)_countof(rtp), &pt)) {
                track->type = toTrackType(type);
            }
            break;
        }
        case 'a': {
            string attr = findSubString(opt_val.data(), nullptr, ":");
            if (attr.empty()) {
                track->map_attr.emplace(opt_val, "");
            }
            else {
                track->map_attr.emplace(attr, findSubString(opt_val.data(), ":", nullptr));
            }
            break;
        }
        }
    }

    for (auto& track_ptr : m_vecTrack) {
        auto& track = *track_ptr;
        auto it = track.map_attr.find("control");
        if (it != track.map_attr.end()) {
            track.szControl = it->second;
        }
    }
}

SdpParser::~SdpParser()
{
    for (auto one : m_vecTrack)
        delete one;
    m_vecTrack.clear();
}


SdpTrack* SdpParser::GetTrack(ETrackType type) const
{
    for (auto& track : m_vecTrack) {
        if (track->type == type) {
            return track;
        }
    }
    return nullptr;
}

std::vector<SdpTrack*> SdpParser::GetAvailableTrack() const
{
    vector<SdpTrack*> ret;
    bool audio_added = false;
    bool video_added = false;
    for (auto& track : m_vecTrack) {
        if (track->type == ETrackType::AUDIO) {
            if (!audio_added) {
                ret.emplace_back(track);
                audio_added = true;
            }
            continue;
        }

        if (track->type == ETrackType::VIDEO) {
            if (!video_added) {
                ret.emplace_back(track);
                video_added = true;
            }
            continue;
        }
    }
    return ret;
}

std::map<std::string, std::string> SdpParser::ParseArgs(const string& str, const char* pair_delim, const char* key_delim)
{
    map<string, string> ret;
    auto arg_vec = split(str, pair_delim);
    for (auto& key_val : arg_vec) {
        if (key_val.empty()) {
            // 忽略
            continue;
        }
        auto pos = key_val.find(key_delim);
        if (pos != string::npos) {
            std::string key(key_val, 0, pos);
            std::string val(key_val.substr(pos + strlen(key_delim)));
            trim(key);
            trim(val);
            ret.emplace(std::move(key), std::move(val));
        }
        else {
            trim(key_val);
            if (!key_val.empty()) {
                ret.emplace(std::move(key_val), "");
            }
        }
    }
    return ret;
}

std::string SdpTrack::GetControlUrl(const std::string& szUrl) const
{
    if (szControl.find("://") != string::npos) {
        // 以rtsp://开头
        return szControl;
    }
    return szUrl + "/" + szControl;
}

10.MediaPusherManager.h

#pragma once

#include 
#include 
#include 
#include 

class RtspSession;

//具体推流类
class MediaPusher {
public:
	MediaPusher(const std::string& szUrl, RtspSession* pRtspSession);
	~MediaPusher();
	//添加一个接收端
	void AddRecvier(RtspSession* pRtspSession);
	//删除一个接收端
	void DelRecvier(RtspSession* pRtspSession);
	//发送数据给所有接收端
	bool SendDataToRecvier(const char* pData, size_t uSize);
	//提供给拉流实例，用于获取推流对应的tcp连接中存储的各项信息，比如sdp
	RtspSession* GetRtspSession() const { return m_pRtspSession; }
private:
	std::string m_szUrl;				//推流地址
	RtspSession* m_pRtspSession = nullptr; //推流对应的tcp连接	
	std::list<RtspSession*> m_listRtspSession;
};

//全局媒体推流管理类
class MediaPusherManager {
public:
	static MediaPusherManager& Instance() {
		static MediaPusherManager instance;
		return instance;
	}
	//添加一个推流地址，返回是否添加成功
	bool AddPusher(const std::string& szUrl, RtspSession* pRtspSession);
	//删除一个推流地址，返回是否删除成功
	bool DelPusher(const std::string& szUrl);
	//判断推流地址是否存在
	bool IsExist(const std::string& szUrl);
	//添加一个接收端，返回对应的tcp连接session，失败返回nullptr
	RtspSession* AddRecvier(const std::string& szUrl, RtspSession* pRtspSession);
	//删除一个接收端，返回是否删除成功
	bool DelRecvier(const std::string& szUrl, RtspSession* pRtspSession);
	//发送数据给所有接收端
	bool SendDataToRecvier(const std::string& szUrl, const char* pData, size_t uSize);
private:
	std::map<const std::string, MediaPusher*> m_mapMediaPusher;
	std::mutex m_mtxMediaPusher;
};

11.MediaPusherManager.cpp

#include "MediaPusherManager.h"
#include "RtspSession.h"

bool MediaPusherManager::AddPusher(const std::string& szUrl, RtspSession* pRtspSession)
{
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) == m_mapMediaPusher.end()) {
            m_mapMediaPusher[szUrl] = new MediaPusher(szUrl, pRtspSession);
            return true;
        }
    }
    return false;
}

bool MediaPusherManager::DelPusher(const std::string& szUrl)
{
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            delete m_mapMediaPusher.at(szUrl);
            m_mapMediaPusher.erase(szUrl);
            return true;
        }
    }
    return false;
}

bool MediaPusherManager::IsExist(const std::string& szUrl)
{
    std::lock_guard lck(m_mtxMediaPusher);
    return m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end();
}

RtspSession* MediaPusherManager::AddRecvier(const std::string& szUrl, RtspSession* pRtspSession)
{
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            auto pusher = m_mapMediaPusher.at(szUrl);
            pusher->AddRecvier(pRtspSession);
            return pusher->GetRtspSession();
        }
    }
    return nullptr;
}

bool MediaPusherManager::DelRecvier(const std::string& szUrl, RtspSession* pRtspSession)
{
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            m_mapMediaPusher.at(szUrl)->DelRecvier(pRtspSession);
            return true;
        }
    }
    return false;
}

bool MediaPusherManager::SendDataToRecvier(const std::string& szUrl, const char* pData, size_t uSize)
{
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            m_mapMediaPusher.at(szUrl)->SendDataToRecvier(pData, uSize);
            return true;
        }
    }
    return false;
}

MediaPusher::MediaPusher(const std::string& szUrl, RtspSession* pRtspSession)
    :m_szUrl(szUrl)
    , m_pRtspSession(pRtspSession)
{
}

MediaPusher::~MediaPusher()
{
}

void MediaPusher::AddRecvier(RtspSession* pRtspSession)
{
    m_listRtspSession.push_back(pRtspSession);
}

void MediaPusher::DelRecvier(RtspSession* pRtspSession)
{
    m_listRtspSession.remove(pRtspSession);
}

bool MediaPusher::SendDataToRecvier(const char* pData, size_t uSize)
{
    decltype(m_listRtspSession) removeList;
    for (auto one : m_listRtspSession) {
        if (false == one->SendData(pData, uSize)) {
            removeList.push_back(one);
        }
    }
    for (auto one : removeList) {
        this->DelRecvier(one);
        one->CloseSession();
    }
    return true;
}