如何基于C++ Drogon库实现无阻塞式摄像头实时视频流上传？

问题 #

我需要实现摄像头实时视频流的无阻塞上传，但当前代码会阻塞线程，导致服务无法正常接收其他请求。现有代码如下：

void video::async_video_stream(const HttpRequestPtr &req, std::function<void(const HttpResponsePtr &)> &&callback) const
{
    cap_.open(path);// Open the default camera
    if (!cap_.isOpened()) {
        auto resp_err = HttpResponse::newHttpResponse();
        resp_err->setStatusCode(k500InternalServerError);
        resp_err->setBody("Failed to open video file");
        callback(resp_err);
        return;
    }

    auto sendFrame = [this](ResponseStreamPtr res) -> void {
        double frame_rate = cap_.get(cv::CAP_PROP_FPS);
        int delay         = static_cast<int>(1000.0 / frame_rate);
        while (cap_.isOpened()) {
            cv::Mat frame;
            cap_ >> frame;// Capture a frame
            if (!frame.empty()) {
                std::vector<uchar> img_buffer;
                cv::imencode(".jpg", frame, img_buffer);
                std::string ima_data(img_buffer.begin(), img_buffer.end());
                auto img_size = img_buffer.size();
                std::ostringstream ss;
                ss << "--frame
Content-Type: image/jpeg

";
                ss << ima_data << "
";
                res->send(ss.str());

                std::this_thread::sleep_for(std::chrono::milliseconds(delay));// 10ms
            } else {
                break;
            }
        }
        res->close();
        cap_.release();
    };

    auto resp = HttpResponse::newAsyncStreamResponse(sendFrame);
    resp->addHeader("Content-Type", "multipart/x-mixed-replace;boundary=frame");
    callback(resp);
}

核心问题分析 #

当前代码的sendFrame回调直接在响应流的处理线程中执行while循环，加上std::this_thread::sleep_for会完全阻塞该线程，导致服务的IO线程池被占满，无法处理其他请求。

无阻塞改造方案 #

把视频采集、编码的耗时逻辑从响应线程剥离到后台工作线程，同时优化资源管理和帧率控制：

1. 用线程池执行后台任务 #

借助框架的线程池（比如Drogon的getThreadPool()）把视频循环放到独立线程，避免占用IO线程资源。

2. 线程安全的资源管理 #

用智能指针封装摄像头对象，避免多线程共享成员变量的安全问题；用弱指针监控响应流状态，客户端断开时及时终止循环。

3. 优化帧率控制 #

通过计算每帧处理耗时，仅在必要时sleep，减少线程阻塞时间。

修改后的代码 #

void video::async_video_stream(const HttpRequestPtr &req, std::function<void(const HttpResponsePtr &)> &&callback) const
{
    // 先尝试打开摄像头，失败直接返回错误
    cv::VideoCapture temp_cap(path);
    if (!temp_cap.isOpened()) {
        auto resp_err = HttpResponse::newHttpResponse();
        resp_err->setStatusCode(k500InternalServerError);
        resp_err->setBody("无法打开摄像头");
        callback(resp_err);
        return;
    }
    // 用shared_ptr管理摄像头，确保线程安全访问
    auto cap = std::make_shared<cv::VideoCapture>(std::move(temp_cap));
    double frame_rate = cap->get(cv::CAP_PROP_FPS);
    int delay = static_cast<int>(1000.0 / frame_rate);

    auto resp = HttpResponse::newAsyncStreamResponse([cap, delay](const ResponseStreamPtr &res) {
        // 用weak_ptr监控响应流，避免客户端断开后悬空引用
        auto weak_res = std::weak_ptr<ResponseStream>(res);
        
        // 提交到后台线程池执行视频流循环
        drogon::app().getThreadPool()->enqueue([cap, delay, weak_res]() {
            auto start_time = std::chrono::steady_clock::now();
            while (cap->isOpened()) {
                // 检查客户端是否已断开连接
                auto strong_res = weak_res.lock();
                if (!strong_res || strong_res->isClosed()) {
                    break;
                }

                cv::Mat frame;
                *cap >> frame;
                if (frame.empty()) {
                    break;
                }

                // 编码帧为JPEG
                std::vector<uchar> img_buffer;
                cv::imencode(".jpg", frame, img_buffer);
                std::string img_data(img_buffer.begin(), img_buffer.end());

                // 构造符合规范的multipart响应内容
                std::ostringstream ss;
                ss << "--frame\r\n";
                ss << "Content-Type: image/jpeg\r\n";
                ss << "Content-Length: " << img_buffer.size() << "\r\n\r\n";
                ss << img_data << "\r\n";

                // 异步发送帧数据
                strong_res->send(ss.str());

                // 计算耗时，动态调整sleep时间，减少阻塞
                auto end_time = std::chrono::steady_clock::now();
                auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count();
                if (elapsed < delay) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(delay - elapsed));
                }
                start_time = std::chrono::steady_clock::now();
            }

            // 释放摄像头资源
            cap->release();
            // 确保响应流正确关闭
            auto strong_res = weak_res.lock();
            if (strong_res) {
                strong_res->close();
            }
        });
    });

    resp->addHeader("Content-Type", "multipart/x-mixed-replace;boundary=frame");
    callback(resp);
}

关键改动说明 #

• 线程隔离：视频循环在后台线程池执行，响应线程仅负责初始化响应后立即返回，不被阻塞。
• 资源安全：摄像头对象用shared_ptr封装，避免多线程访问冲突；通过weak_ptr检测客户端连接状态，及时终止循环释放资源。
• 规范兼容：补充Content-Length头，符合HTTP multipart格式要求，避免解析异常。
• 帧率优化：基于实际处理耗时动态调整sleep时长，减少不必要的线程阻塞。

内容的提问来源于stack exchange，提问作者GuardHer