C++流媒体底层架构：从零构建千万级并发直播系统的关键技术与实践jzit

// net/event_loop.h
class EventLoop {
public:
    void Run() {
        while (!stop_) {
            int nfds = epoll_wait(epfd_, events_, kMaxEvents, 5);
            for (int i = 0; i < nfds; ++i) {
                auto* channel = static_cast<Channel*>(events_[i].data.ptr);
                channel->HandleEvent(events_[i].events);
            }
            // 执行定时器（GOP对齐、心跳、码率探测）
            timer_.Tick();
        }
    }
    void AddConnection(int fd, const InetAddr& peer);
private:
    int epfd_;
    epoll_event events_[512];
    TimerWheel timer_;
    std::atomic<bool> stop_;
};

// protocol/rtmp_chunk_decoder.h
class RtmpChunkDecoder {
public:
    // 返回完整消息ID，0表示还未收齐
    uint32_t Feed(const uint8_t* data, size_t len, std::vector<MediaMessage>& out) {
        while (offset_ < len) {
            if (state_ == kReadBasicHeader) {
                basic_header_.Parse(data[offset_++]);
                if (basic_header_.fmt == 0) state_ = kReadMsgHeader7;
                else if (basic_header_.fmt == 1) state_ = kReadMsgHeader3;
                // ...
            }
            // 处理扩展时间戳 (CSID > 2)
            if (basic_header_.has_extended_ts) {
                extended_ts_ = (data[offset_+3]<<24) | ...;
                offset_ += 4;
            }
            // 拼接payload到对应chunk stream的buffer
            auto& cs = chunk_streams_[basic_header_.csid];
            size_t copy_len = std::min(len - offset_, cs.remaining);
            cs.buffer.insert(cs.buffer.end(), data + offset_, data + offset_ + copy_len);
            offset_ += copy_len;
            if (cs.remaining == 0) {
                out.emplace_back(std::move(cs.buffer), cs.msg_type, cs.timestamp);
                cs.Reset();
            }
        }
        return total_messages_;
    }
private:
    struct ChunkStream {
        std::vector<uint8_t> buffer;
        uint32_t remaining;
        uint32_t msg_type;
        uint32_t timestamp;
    };
    std::unordered_map<uint32_t, ChunkStream> chunk_streams_;  // CSID -> stream
    uint8_t state_;
    uint32_t offset_;
};

// memory/fixed_pool.h
template<size_t BlockSize>
class FixedPool {
public:
    void* Alloc() {
        if (free_list_.empty()) {
            // 向OS申请2MB大页
            char* new_block = (char*)mmap(nullptr, 2*1024*1024, PROT_READ|PROT_WRITE,
                                          MAP_PRIVATE|MAP_ANONYMOUS|MAP_HUGETLB, -1, 0);
            // 切割成BlockSize大小的链表
            for (size_t i=0; i< (2*1024*1024)/BlockSize; ++i) {
                free_list_.push_back(new_block + i*BlockSize);
            }
        }
        auto* ptr = free_list_.back(); free_list_.pop_back();
        return ptr;
    }
    void Dealloc(void* p) { free_list_.push_back((char*)p); }
private:
    std::vector<char*> free_list_;  // 无锁栈 (实际用moodycamel::ConcurrentQueue)
};

class MediaFrame {
    std::shared_ptr<BufferRef> data;  // 引用计数指向池中内存
    uint32_t ref_count;
};

// send/zero_copy_sender.cc
void ZeroCopySender::SendFrame(std::shared_ptr<MediaFrame> frame) {
    struct iovec iov[2];
    iov[0].iov_base = rtmp_header_buf_;   // 预置头部
    iov[0].iov_len = RTMP_HEADER_SIZE;
    iov[1].iov_base = frame->data();
    iov[1].iov_len = frame->size();

    // 使用io_uring提供的固定文件描述符（注册socket）
    struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
    io_uring_prep_writev(sqe, sock_fd_, iov, 2, 0);
    io_uring_sqe_set_flags(sqe, IOSQE_FIXED_FILE);
    io_uring_submit(&ring_);
    // 异步回调中更新发送窗口和码率统计
}

// cache/gop_cache.h
class GopCache {
public:
    void AppendFrame(std::shared_ptr<MediaFrame> frame) {
        if (frame->is_keyframe) {
            // 开启新GOP，将旧GOP放入LRU
            if (gop_count_ >= max_gop_) pop_front();
            current_gop_.clear();
        }
        current_gop_.push_back(frame);
    }
    std::vector<std::shared_ptr<MediaFrame>> GetLastGop() {
        return (gop_list_.empty() ? current_gop_ : gop_list_.back());
    }
private:
    std::deque<std::vector<std::shared_ptr<MediaFrame>>> gop_list_;
    std::vector<std::shared_ptr<MediaFrame>> current_gop_;
    size_t max_gop_ = 3;
};

// abr/rate_controller.cc
uint32_t RateController::SelectLevel(uint32_t current_buffer_ms, uint32_t network_kbps) {
    // 使用MPC（模型预测控制）选择最优码率
    auto& profile = profiles_[current_profile_index_];
    if (network_kbps < profile.min_bitrate) {
        return DownShift();  // 降档
    } else if (current_buffer_ms > kHighBufferThreshold) {
        return UpShift();    // 升档
    }
    return current_profile_index_;
}

class PipelineManager {
    void Switchover() {
        std::lock_guard<std::mutex> lock(mtx_);
        auto new_active = standby_->DrainAndReset();  // 新active必须清空旧状态
        std::swap(active_, standby_);
        // 通知所有下游session重置解码器（发送SPS/PPS）
        for (auto& session : sessions_) {
            session->SendDecoderConfigurationRecord();
        }
    }
};