Simply Patrick Hopes can always go up, tears can only come down.

架設一台網路監控攝影機（IP Camera）或即時影像串流伺服器，是許多人在學習物聯網（IoT）與電腦視覺時的經典入門專案。但在實務上，最常遇到的技術瓶頸莫過於：硬體影像解碼與電腦視覺（CV）影像處理解析度太高，一不小心就把 Web 伺服器的非同步事件迴圈（Event Loop）給卡死。

在這次的 Rust 52 Projects 挑戰中，我用 Rust + OpenCV 5 + Axum (v0.7) + Tokio 打造了一個能跨平台運行於 Windows (MSVC) 與 Linux / 樹莓派 4（Raspberry Pi 4 ARM64）的高效能 IP Camera 視訊串流伺服器：ip-camera。

這篇文章將全面拆解其架構設計，包含硬體擷取與 Web 異步解耦、Worker 執行緒 JPEG 壓縮、tokio::sync::watch 廣播通知、OpenCV 5 圖像處理管線，以及無鏡頭環境下的雷達模擬備援機制！

專案亮點與技術特色

• 獨立多執行緒擷取迴圈 (spawn_blocking)：硬體影像擷取、CV 圖形繪製與 JPEG 壓縮隔離在獨立 Worker 執行緒，保持 Axum/Tokio 非同步 Web 事件迴圈極速回應。
• OpenCV 5 即時視覺管線：
  • EMA 平滑即時 FPS 計數器。
  • 高精度微秒級時間戳記 Overlay。
  • 互動式 HUD 掃描框：含四角綠色標記、動態雷射掃描紅線與特定 ROI 區域動態高斯模糊（Gaussian Blur）。
• Worker 執行緒內建 JPEG 壓縮：在背景 Worker 執行緒直接完成 cv::Mat 到 JPEG 的 imencode 壓縮，Web 串流任務只需廣播 Raw Bytes，大幅降低多用戶同時連線時的 CPU 開銷。
• tokio::sync::watch 異步喚醒機制：Web 串流任務在沒有新影格時完全處於休眠狀態（Zero Overhead），唯有 Worker 產出新影格時才被非同步喚醒發送，防止 Busy Polling。
• 韌性離線備援 (Mock Radar Scope)：當無實體 Camera 連線時（如 Headless VM、CI 環境），自動切換至動態旋轉雷達掃描儀模擬視訊源。
• 現代 Glassmorphism Web 儀表板：內建暗黑玻璃擬物化 Web 介面，支援全螢幕、快照截圖與播放/暫停控制。

系統串流架構：視訊處理與 Web 伺服器解耦

為避免耗時的 OpenCV 矩陣運算與 JPEG 壓縮阻塞 Axum 處理 HTTP 請求的 Tokio Worker Threads，整個系統在架構上進行了徹底的分工解耦：

核心技術一：非同步解耦與 tokio::task::spawn_blocking

在 src/main.rs 中，我們建立了一個共享的線程安全記憶體空間 Arc<Mutex<Vec<u8>>> 保存最新一幀的 JPEG 壓縮 Byte Array，以及一個 watch::channel 負責通知廣播。

接著，透過 tokio::task::spawn_blocking 將 OpenCV 的硬體讀取與圖像處理解耦出去：

// Shared thread-safe storage for the latest encoded frame.
let frame_state = Arc::new(Mutex::new(initial_jpeg.to_vec()));
let frame_state_clone = Arc::clone(&frame_state);

// Watch channel to notify web clients of new frames.
let (watch_tx, watch_rx) = watch::channel(0u64);

// 將硬體 VideoCapture 迴圈隔離在獨立的 blocking worker 執行緒
let camera_index = args.camera_index;
let filters = args.filter.clone();
tokio::task::spawn_blocking(move || {
    if let Err(e) = camera::run_capture_loop(camera_index, filters, frame_state_clone, watch_tx) {
        error!("Capture loop exited with error: {:?}", e);
    }
});

💡 架構深挖：為什麼不直接用 Channel 傳送 JPG 影像？

許多人在設計視訊串流時，第一直覺是建立一個 mpsc::channel 或 broadcast::channel，將壓縮好的 Vec<u8> JPG 影像直接扔進 Channel 送給 Web 伺服器。但在高效能 IP Camera 的情境下，我們選擇了 Arc<Mutex<Vec<u8>>> (共享記憶體) + tokio::sync::watch (訊號廣播) 的組合，主要考量如下：

1. 保證「永遠只看最新影格 (Latest-Frame-Only)」的即時性： IP 監控攝影機的核心要求是極低延遲與即時性。使用者想看的是「此刻發生了什麼」，而不是 3 秒前積壓在佇列裡的歷史影格。 若使用傳統管道（Channel Queue），當某些 Web 客戶端網路較慢（例如用手機 3G 觀看）時，Channel 內部會積壓大量未消費的 JPG 影格，導致畫面延遲越來越大（Lag）且記憶體不斷暴增。 而 watch 頻道搭配共享記憶體具有天然的 最新狀態覆寫 (Single-Slot Overwrite) 語意：慢速客戶端被喚醒時，永遠只會讀取到當前最新的那張 JPG 影格，自動跳過中間沒來得及發送的影格，實現零積壓、零延遲！

2. 避免多連線時的記憶體重複複製 (Zero Fan-Out Memory Inflation)： 若透過 broadcast channel 發送 Vec<u8> 影像，每當有 $N$ 個客戶端連線時，系統就必須將圖像 Payload 複製 $N$ 份或處理複雜的共享記憶體佇列。 採用 Arc<Mutex<Vec<u8>>> 後，無論是 0 個還是 100 個客戶端連線，背景 Worker 執行緒永遠只執行一次 JPEG 壓縮與單一記憶體覆寫，將記憶體開銷牢牢鎖定在最小的單影格大小。

3. 無人觀看時的極致零開銷 (Zero-Cost Idle)： 當無人連線觀看視訊時，Worker 執行緒更新 Arc<Mutex> 並呼叫 watch_tx.send(frame_id) 僅會更新一個 $u64$ 的 Frame Counter 號碼牌，完全不會產生任何排隊訊息開銷或記憶體洩漏風險。

核心技術二：OpenCV 5 圖像處理管線與 EMA FPS 計算

在 src/camera.rs 的擷取迴圈中，我們不僅處理影像，還利用 Exponential Moving Average (EMA) 演算法計算平滑的即時 FPS，避免數據劇烈跳動：

// 計算即時動態 FPS
frame_count += 1;
let now = Instant::now();
let delta = now.duration_since(last_fps_time).as_secs_f64();
last_fps_time = now;
if delta > 0.0 {
    let instant_fps = 1.0 / delta;
    // 使用 EMA 演算法平滑化 FPS 數值
    fps = fps * 0.95 + instant_fps * 0.05;
}

HUD 掃描框與動態雷射光束

在 CV 濾波器管線中，我們能隨意疊加高斯模糊 (Blur)、Canny 邊緣檢測 (Canny)、灰階 (Grayscale)、色彩反轉 (Invert)，或者啟動一個帶有紅光雷射掃描與綠色 HUD 角落標記的 Scanner 模式：

FilterConfig::Scanner { margin } => {
    if let Ok(size) = current_frame.size() {
        let m = *margin;
        let box_w = std::cmp::max(1, size.width - m * 2);
        let box_h = std::cmp::max(1, size.height - m * 2);
        let roi_rect = Rect::new(m, m, box_w, box_h);

        // 畫出 HUD 外框與綠色四角標記
        let _ = imgproc::rectangle(&mut current_frame, roi_rect, Scalar::new(99.0, 102.0, 241.0, 0.0), 2, imgproc::LINE_AA, 0);

        // 計算雷射紅線上下掃描的位置
        let scan_period = 100;
        let scan_pos = (frame_count % scan_period) as i32;
        let scan_y = m + (scan_pos * box_h / scan_period as i32);
        let _ = imgproc::line(
            &mut current_frame,
            Point::new(m + 2, scan_y),
            Point::new(m + box_w - 2, scan_y),
            Scalar::new(0.0, 0.0, 255.0, 0.0), // 雷射紅光
            2,
            imgproc::LINE_AA,
            0,
        );
    }
}

在 Worker 執行緒預先完成 JPEG 壓縮

為了避免 10 個 Web 客戶端連線時，伺服器必須執行 10 次重複的 JPEG 壓縮運算，我們直接在 Capture 迴圈最後將 Mat 壓縮成 85% 品質的 JPEG Byte 陣列：

let mut jpeg_buf = Vector::<u8>::new();
let mut encode_params = Vector::<i32>::new();
encode_params.push(imgcodecs::IMWRITE_JPEG_QUALITY);
encode_params.push(85); // 85% 品質：兼顧畫質與網路頻寬

imgcodecs::imencode(".jpg", &frame, &mut jpeg_buf, &encode_params)?;

// 將 compressed bytes 寫入共享記憶體，並發送 watch 廣播
{
    let mut lock = frame_state.lock().unwrap();
    *lock = jpeg_buf.to_vec();
}
let _ = watch_tx.send(watch_tx.borrow().wrapping_add(1));

核心技術三：Axum 0.7 異步 MJPEG 串流與 stream::unfold

多媒體 MJPEG (Motion JPEG) 串流採用 HTTP 標準的 multipart/x-mixed-replace; boundary=frame 標頭。每個 Chunk 由 --frame 分界，隨後接上 Content-Type: image/jpeg 與 raw bytes。

在 src/handlers.rs 中，我們透過 futures_util::stream::unfold 搭配 watch_rx.changed().await 構造出極致高效的非同步串流 Response：

pub async fn stream_handler(State(state): State<AppState>) -> impl IntoResponse {
    let rx = state.watch_rx.clone();
    let frame_state = state.frame_state.clone();

    // 利用 stream::unfold 打造非同步影格串流
    let stream = stream::unfold((true, rx, frame_state), move |(is_first, mut rx, frame_state)| async move {
        if !is_first {
            // 沒有新影格時，Task 在此處非同步休眠，不消耗 CPU！
            if rx.changed().await.is_err() {
                return None; // Sender dropped
            }
        }

        // 從共享記憶體讀取最新的 JPEG Bytes
        let jpeg_bytes = {
            let lock = frame_state.lock().unwrap();
            lock.clone()
        };

        // 封裝 multipart/x-mixed-replace boundary 封包
        let header = format!(
            "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: {}\r\n\r\n",
            jpeg_bytes.len()
        );
        let mut body_bytes = Vec::new();
        body_bytes.extend_from_slice(header.as_bytes());
        body_bytes.extend_from_slice(&jpeg_bytes);
        body_bytes.extend_from_slice(b"\r\n");

        Some((
            Ok::<Bytes, std::io::Error>(Bytes::from(body_bytes)),
            (false, rx, frame_state),
        ))
    });

    let body = Body::from_stream(stream);
    let mut headers = HeaderMap::new();
    headers.insert(
        CONTENT_TYPE,
        HeaderValue::from_static("multipart/x-mixed-replace; boundary=frame"),
    );

    (headers, body)
}

💡 深入剖析 stream::unfold：如何用數行程式碼生成無限非同步串流？

在函數式程式設計（Functional Programming）與 Rust 非同步生態集中，unfold 是 fold 的對偶（Dual）操作：fold 是將一個集合「坍縮/歸約」成單一數值，而 unfold 則是從一個初始狀態種子（Initial State Seed）開始，依序「展開」產生一個無限或有限的非同步 Stream。

stream::unfold 的運算模型如下：

在 stream_handler 中，我們傳入初始狀態 Tuple (is_first: true, rx, frame_state)：

1. 第 1 次迭代 (is_first = true)： 跳過 rx.changed().await 等待，直接從共享記憶體讀取當前最新的 JPEG Byte Array，組裝出首張 multipart 封包。回傳 Some((Ok(Bytes), (false, rx, frame_state)))。這使得瀏覽器一開啟網頁就能瞬間載入第一張畫面（Zero Delay First Frame）！
2. 後續迭代 (is_first = false)： 執行 rx.changed().await。此時 Tokio Task 會進入完全休眠狀態（0 CPU 消耗）。當背景 Capture Worker 發送新影格訊號時，Task 被喚醒，讀取最新 JPG bytes，並再次產出 Some((Ok(Bytes), (false, rx, frame_state)))。
3. 串流終止 (Sender dropped)： 若背景擷取執行緒關閉，rx.changed().await 回傳 Err(_)，閉包回傳 None，stream::unfold 便會優雅地宣告 Stream 結束，通知 Axum 與 TCP Socket 關閉連線。

為什麼選擇 stream::unfold 而非手動實作 Stream Trait？ 若要手動為自訂 Struct 實作 futures_util::Stream，我們必須編寫 poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Item>>，手動處理 Pin 指標、 unsafe 轉換、狀態機以及 Waker 的註冊與喚醒機制。 stream::unfold 將這一切繁瑣的底層 Async State Machine 封裝起來，讓我們能直接在 async move 區塊中使用標準的 .await 語法，極致優雅地將異步頻道通知轉譯為 Axum 相容的 axum::body::Body::from_stream！

當網頁瀏覽器打開 http://localhost:8080/stream 時，無需任何額外的 JavaScript 播放器，<img> 標籤就能原生、滑順地播放即時視訊串流！

韌性離線備援：模擬雷達儀 (Mock Radar Scope)

在 Headless VM 或 CI/CD 環境下測試時，往往沒有實體 Webcam。專案設計了自動備援機制：當 videoio::VideoCapture 打開失敗時，自動進入 is_simulated = true 模式，繪製動態雷達掃描儀：

// 繪製動態旋轉雷達綠光束與脈衝目標
let center = Point::new(size.width / 2, size.height / 2);
let angle = (frame_count as f64 * 0.05) % (2.0 * std::f64::consts::PI);
let radius = 120.0;
let radar_end = Point::new(
    (center.x as f64 + radius * angle.cos()) as i32,
    (center.y as f64 + radius * angle.sin()) as i32,
);
imgproc::line(&mut frame, center, radar_end, Scalar::new(0.0, 255.0, 0.0, 0.0), 1, imgproc::LINE_AA, 0)?;

imgproc::put_text(&mut frame, "⚠️ CAMERA OFFLINE - RUNNING IN SIMULATED MODE", Point::new(20, size.height - 20), ...)?;

這保證了伺服器在任何環境下都能正常啟動並提供穩定的串流服務。

跨平台編譯與運行指南

1. Windows (MSVC) 環境

專案內附 build.bat 腳本，自動設定 OPENCV_LINK_PATHS、OPENCV_INCLUDE_PATHS 與 runtime DLL PATH：

# 編譯並運行 Release 版本
.\build.bat run --release

2. Linux & 樹莓派 4 (Raspberry Pi 4 ARM64)

在 Debian / Raspberry Pi OS 上，只要安裝 clang 與 pkg-config 即可輕鬆編譯：

sudo apt update && sudo apt install -y build-essential clang libclang-dev pkg-config
cargo run --release

實務討論與技術體會 (Technical Trade-offs)

作為一個教學與實驗導向的專案，我們也必須坦實討論 MJPEG 串流的技術權衡：

1. MJPEG vs RTSP/H.264/H.265：
   • 優點：MJPEG 不需要複雜的跨平台 H.264 硬體編碼器（如 NVENC 或 QuickSync），相容性極佳，所有瀏覽器用普通 <img> 標籤就能播放。
   • 缺點：因為每一影格都是完整的 JPEG 圖片，缺乏 Frame 之間的 Intra-frame 壓縮，頻寬佔用較大（640x360@30fps 約需 2~4 Mbps）。
2. 記憶體拷貝優化空間： 目前 shared state 採用 Mutex<Vec<u8>> 搭配 lock().clone()，在極高並發（如數百個客戶端）時可進一步改用 bytes::Bytes 實現零拷貝（Zero-Copy）廣播。

但作為 Rust 52 Projects 挑戰，這個專案完美展示了如何優雅地組合 Tokio 非同步生態系、Axum 7 路由 與 OpenCV C++ 綁定，打造出兼具效能與韌性的視訊服務！

學到的 Rust 關鍵技術

結語

ip-camera 專案展示了 Rust 在高效能視訊串流與電腦視覺領域的強大能力。從非同步解耦到跨平台（Windows / 樹莓派）支援，整個設計簡潔而堅固。

歡迎前往專案 Repo 查看程式碼並親自試跑！

參考資源

• ip-camera 原始碼 Repo — 本專案完整 Rust 程式碼
• Axum Documentation — Axum Web 框架官方文件
• OpenCV Rust Bindings — OpenCV 的 Rust 綁定庫