GB28181协议客户端开发：SIP与RTP媒体流处理

1. 国标GB28181客户端开发概述

在安防监控领域，GB28181协议已经成为视频监控设备互联互通的国家标准协议。作为一名长期从事视频监控系统开发的工程师，我将分享如何使用C/C++实现一个完整的GB28181客户端程序，用于拉取IPC相机的媒体流。

这个客户端程序的核心功能包括：

• 基于SIP协议实现与IPC相机的信令交互
• 基于UDP协议接收RTP封装的媒体流
• 解析PS（Program Stream）格式的音视频数据
• 提供裸码流（H.264/H.265/AAC）回调接口

程序架构上分为两个主要模块：

1. SIP服务模块：处理设备注册、会话建立和保活等信令交互
2. UDP服务模块：接收和解析IPC相机发送的RTP媒体流

2. SIP服务实现详解

2.1 SIP协议栈选型与初始化

在实现GB28181客户端时，我们选择了eXosip2作为SIP协议栈。eXosip2是osip2协议栈的扩展实现，提供了更高级的API接口，特别适合快速开发SIP应用。

初始化SIP服务的核心代码如下：

cpp复制int init_sip_server(const char* local_ip, int local_port) {
    // 初始化eXosip上下文
    g_excontext = eXosip_malloc();
    if (!g_excontext) {
        std::cerr << "eXosip初始化失败" << std::endl;
        return -1;
    }

    // 配置传输协议（UDP）
    eXosip_set_user_agent(g_excontext, "GB28181 Client");
    if (eXosip_init(g_excontext) != 0) {
        std::cerr << "eXosip初始化失败" << std::endl;
        return -1;
    }

    // 绑定本地IP和端口
    if (eXosip_listen_addr(g_excontext, IPPROTO_UDP, local_ip, local_port, AF_INET, 0) != 0) {
        std::cerr << "无法绑定到指定地址" << std::endl;
        eXosip_quit(g_excontext);
        return -1;
    }

    // 启动事件处理线程
    std::thread eventThread(exosip_event_loop);
    eventThread.detach();

    return 0;
}

注意：在实际项目中，建议将SIP服务封装为一个独立的类，管理eXosip上下文和相关资源，避免全局变量的使用。

2.2 设备注册流程实现

GB28181协议中，IPC相机作为SIP客户端会主动向我们的客户端程序（作为SIP服务器）发起注册请求。注册流程的核心事件处理如下：

cpp复制void handle_exosip_event(eXosip_event_t* event) {
    if (!event) return;

    switch (event->type) {
        case EXOSIP_REGISTRATION_NEW:  // 新注册请求
            on_ipc_register(g_excontext, event->type, event);
            break;
        // 其他事件处理...
    }
    eXosip_event_free(event);
}

void on_ipc_register(eXosip_t* ex, int type, eXosip_event_t* event) {
    // 解析IPC设备信息
    osip_uri_t* from_uri = nullptr;
    osip_from_get_url(event->request->from, &from_uri);
    
    // 保存IPC信息
    g_ipc_uri = osip_uri_to_str(from_uri);
    g_ipc_ip = inet_ntoa(((struct sockaddr_in*)event->src_addr)->sin_addr);

    // 构造200 OK响应
    osip_message_t* response = nullptr;
    eXosip_message_build_answer(ex, event->tid, 200, &response);
    
    // 设置Expires头
    osip_header_t* expires = nullptr;
    osip_message_get_expires(event->request, 0, &expires);
    if (expires) {
        osip_message_set_expires(response, expires->hvalue);
    }
    
    eXosip_message_send_answer(ex, event->tid, 200, response);
}

2.3 媒体流请求(INVITE)流程

设备注册成功后，客户端需要主动发起INVITE请求来建立媒体会话：

cpp复制int send_invite_to_ipc(const char* ipc_ip, const char* local_ip, int rtp_port) {
    // 构造SDP消息体
    std::string sdp = construct_sdp(local_ip, rtp_port);
    
    // 创建INVITE请求
    eXosip_lock(g_excontext);
    int call_id = eXosip_call_build_initial_invite(
        g_excontext, nullptr, 
        ("sip:" + std::string(ipc_ip)).c_str(), 
        nullptr, nullptr, 
        sdp.c_str());
    
    // 发送INVITE请求
    int ret = eXosip_call_send_initial_invite(g_excontext, call_id);
    eXosip_unlock(g_excontext);
    
    return ret;
}

std::string construct_sdp(const char* local_ip, int rtp_port) {
    std::ostringstream oss;
    oss << "v=0\r\n"
        << "o=" << local_ip << " 0 0 IN IP4 " << local_ip << "\r\n"
        << "s=Play\r\n"
        << "c=IN IP4 " << local_ip << "\r\n"
        << "t=0 0\r\n"
        << "m=video " << rtp_port << " RTP/AVP 96\r\n"
        << "a=recvonly\r\n"
        << "a=rtpmap:96 PS/90000\r\n";
    return oss.str();
}

2.4 保活机制实现

GB28181要求设备通过定期发送MESSAGE消息保持会话活跃：

cpp复制void handle_keepalive(eXosip_event_t* event) {
    if (event->request && 
        osip_strcasecmp(event->request->sip_method, "MESSAGE") == 0) {
        // 解析消息内容（XML格式）
        osip_body_t* body = nullptr;
        osip_message_get_body(event->request, 0, &body);
        
        // 构造200 OK响应
        osip_message_t* response = nullptr;
        eXosip_message_build_answer(g_excontext, event->tid, 200, &response);
        eXosip_message_send_answer(g_excontext, event->tid, 200, response);
    }
}

3. UDP媒体流接收与处理

3.1 UDP服务端实现

我们实现了一个UDP服务端类来接收RTP封装的媒体流：

cpp复制class UdpPsServer {
public:
    bool initServer(int port, const std::string& ip = "0.0.0.0") {
        // 创建UDP套接字
        m_sockfd = socket(AF_INET, SOCK_DGRAM, 0);
        if (m_sockfd == INVALID_SOCKET) {
            std::cerr << "创建Socket失败" << std::endl;
            return false;
        }

        // 配置服务端地址
        memset(&m_serverAddr, 0, sizeof(m_serverAddr));
        m_serverAddr.sin_family = AF_INET;
        m_serverAddr.sin_port = htons(port);
        m_serverAddr.sin_addr.s_addr = inet_addr(ip.c_str());

        // 绑定地址
        if (bind(m_sockfd, (sockaddr*)&m_serverAddr, sizeof(m_serverAddr)) == SOCKET_ERROR) {
            std::cerr << "绑定地址失败" << std::endl;
            closeSocket();
            return false;
        }

        m_isInit = true;
        return true;
    }

    int recvData(char* recvBuf, int bufSize, std::string& clientIp, int& clientPort) {
        sockaddr_in clientAddr;
        socklen_t addrLen = sizeof(clientAddr);
        
        int recvLen = recvfrom(m_sockfd, recvBuf, bufSize, 0,
                             (sockaddr*)&clientAddr, &addrLen);
        if (recvLen > 0) {
            clientIp = inet_ntoa(clientAddr.sin_addr);
            clientPort = ntohs(clientAddr.sin_port);
        }
        return recvLen;
    }
};

3.2 PS流解析器实现

PS（Program Stream）是GB28181中常用的封装格式，我们需要解析PS流获取音视频裸码流：

cpp复制class PsParser {
public:
    void setCallback(std::function<void(uint8_t*, size_t, int)> cb) {
        m_callback = cb;
    }

    void parseRtpPsData(const uint8_t* data, size_t len) {
        // 1. 解析RTP头
        RtpHeader* rtp = (RtpHeader*)data;
        uint16_t seq = ntohs(rtp->seq);
        uint32_t timestamp = ntohl(rtp->timestamp);
        
        // 2. 处理PS流
        const uint8_t* psData = data + sizeof(RtpHeader);
        size_t psLen = len - sizeof(RtpHeader);
        
        // 3. 解析PS包头
        if (psLen > 4 && psData[0] == 0x00 && psData[1] == 0x00 && 
            psData[2] == 0x01 && psData[3] == 0xBA) {
            // 这是一个PS包头
            parsePsHeader(psData, psLen);
        } else {
            // 这是PS包体
            parsePsPayload(psData, psLen);
        }
    }

private:
    void parsePsHeader(const uint8_t* data, size_t len) {
        // 解析系统头、映射流等
        // ...
    }

    void parsePsPayload(const uint8_t* data, size_t len) {
        // 解析PES包，提取音视频帧
        // ...
        if (m_callback) {
            m_callback(frameData, frameLen, frameType);
        }
    }
};

3.3 媒体流处理主循环

cpp复制void dealReceiveStreamEx() {
    UdpPsServer udpServer;
    PsParser psParser;
    
    // 设置回调函数
    psParser.setCallback([](uint8_t* data, size_t len, int type) {
        if (type == 0) { // 视频帧
            std::cout << "收到视频帧，大小：" << len << "字节" << std::endl;
            saveToFile("video.h264", data, len);
        } else if (type == 1) { // 音频帧
            std::cout << "收到音频帧，大小：" << len << "字节" << std::endl;
            saveToFile("audio.aac", data, len);
        }
    });

    // 初始化UDP服务
    if (!udpServer.initServer(9000)) {
        return;
    }

    // 主接收循环
    char buffer[1024 * 1024];
    std::string clientIp;
    int clientPort;
    
    while (true) {
        int len = udpServer.recvData(buffer, sizeof(buffer), clientIp, clientPort);
        if (len > 0) {
            psParser.parseRtpPsData((const uint8_t*)buffer, len);
        }
    }
}

4. 关键问题与解决方案

4.1 时间戳处理问题

在接收RTP流时，我们发现时间戳处理不当会导致音视频不同步。解决方案是：

1. 正确解析RTP头中的时间戳字段
2. 根据时间戳计算PTS（Presentation Time Stamp）
3. 对音视频帧进行缓冲和同步处理

cpp复制void handleRtpPacket(const RtpHeader* rtp, const uint8_t* payload, size_t len) {
    uint32_t rtpTimestamp = ntohl(rtp->timestamp);
    uint32_t pts = convertRtpToPts(rtpTimestamp);
    
    // 根据帧类型处理
    if (isVideoFrame(payload)) {
        m_videoBuffer.addFrame(payload, len, pts);
    } else if (isAudioFrame(payload)) {
        m_audioBuffer.addFrame(payload, len, pts);
    }
    
    // 同步处理
    synchronizeAvFrames();
}

4.2 PS流解析异常处理

PS流解析过程中可能会遇到各种异常情况，我们增加了以下保护措施：

1. 数据完整性检查
2. 错误恢复机制
3. 日志记录

cpp复制void parsePsPayload(const uint8_t* data, size_t len) {
    // 检查起始码
    if (len < 4 || data[0] != 0x00 || data[1] != 0x00 || 
        data[2] != 0x01) {
        logError("无效的PES起始码");
        return;
    }
    
    // 解析PES头
    uint8_t streamId = data[3];
    size_t pesHeaderLen = 6; // 基本头长度
    
    // 检查PES头长度
    if (len < pesHeaderLen) {
        logError("PES包过短");
        return;
    }
    
    // 解析PTS/DTS
    uint64_t pts = 0, dts = 0;
    if (data[7] & 0x80) { // PTS标志
        pts = parseTimestamp(data + 9);
        pesHeaderLen += 5;
    }
    
    // 检查有效载荷长度
    if (len <= pesHeaderLen) {
        logError("无有效载荷");
        return;
    }
    
    // 处理有效载荷
    const uint8_t* payload = data + pesHeaderLen;
    size_t payloadLen = len - pesHeaderLen;
    
    // ... 进一步处理
}

4.3 会话终止流程

当IPC主动发送BYE请求时，需要正确关闭会话：

cpp复制void handle_bye(eXosip_event_t* event) {
    // 发送200 OK响应
    osip_message_t* response = nullptr;
    eXosip_message_build_answer(g_excontext, event->tid, 200, &response);
    eXosip_message_send_answer(g_excontext, event->tid, 200, response);
    
    // 释放媒体资源
    release_media_resources();
    
    // 重置会话状态
    reset_session_state();
}

5. 性能优化与扩展

5.1 多线程处理模型

为提高性能，我们采用了多线程模型：

1. 主线程：处理SIP信令
2. 接收线程：处理UDP媒体流
3. 解析线程：解析PS流
4. 处理线程：处理音视频数据

cpp复制void start_media_pipeline() {
    // 创建线程池
    m_threadPool.resize(4);
    
    // SIP事件处理线程
    m_threadPool[0] = std::thread(&GbClient::sipEventLoop, this);
    
    // UDP接收线程
    m_threadPool[1] = std::thread(&GbClient::udpReceiveLoop, this);
    
    // PS解析线程
    m_threadPool[2] = std::thread(&GbClient::psParseLoop, this);
    
    // 数据处理线程
    m_threadPool[3] = std::thread(&GbClient::dataProcessLoop, this);
}

5.2 支持H.265编码

为支持H.265编码，我们扩展了PS解析器：

cpp复制void parseVideoFrame(const uint8_t* data, size_t len) {
    // 检查编码类型
    if (isH264(data)) {
        parseH264Frame(data, len);
    } else if (isH265(data)) {
        parseH265Frame(data, len);
    } else {
        logError("未知的视频编码类型");
    }
}

bool isH265(const uint8_t* data) {
    // 检查H.265起始码和NAL单元类型
    return (len > 4 && data[0] == 0x00 && data[1] == 0x00 && 
            data[2] == 0x01 && ((data[3] & 0x7E) >> 1) == 32);
}

5.3 断线重连机制

为提高稳定性，实现了断线自动重连：

cpp复制void check_connection() {
    if (time(nullptr) - m_lastKeepalive > TIMEOUT_THRESHOLD) {
        logWarning("连接超时，尝试重连...");
        
        // 释放现有资源
        release_resources();
        
        // 重新初始化
        if (init_sip_server(m_localIp.c_str(), m_localPort) != 0) {
            logError("重连失败");
            return;
        }
        
        // 重新发送INVITE
        send_invite_to_ipc(m_ipcIp.c_str(), m_localIp.c_str(), m_rtpPort);
    }
}

在实际项目中，这个GB28181客户端程序已经稳定运行在各种安防监控场景中。通过合理的架构设计和细致的异常处理，它能够可靠地接收和处理来自不同厂商IPC相机的媒体流。对于希望深入了解GB28181协议实现的开发者，建议从SIP协议基础开始，逐步扩展到媒体流处理部分，同时注意协议中的各种细节要求。