AIORTC 实现桌面实时屏幕流媒体传输

本文详解如何用 aiortc 替代静态视频文件，结合 mss 库实现低延迟、高帧率的本地桌面实时 WebRTC 流媒体推流，涵盖自定义 VideoStreamTrack、线程安全帧队列、时间戳同步等关键实践。

在基于 aiortc 的 WebRTC 服务中，MediaPlayer 仅适用于预录制的音视频文件（如 video.mp4），无法满足实时屏幕共享需求。要将用户当前桌面画面作为实时视频源推送到浏览器客户端，必须实现一个符合 aiortc 接口规范的自定义 VideoStreamTrack，并集成高效、轻量的屏幕捕获方案。mss 是 Python 中性能优异的纯 Python 屏幕捕获库，支持跨平台、无依赖、高帧率（可达 30–60 FPS），是理想选择。

以下是一个完整、可运行的 ScreenCapturing 类实现，它继承 VideoStreamTrack，通过后台线程持续抓取屏幕，并利用线程安全队列（queue.Queue）向主线程异步提供帧：

from aiortc import VideoStreamTrack
from av import VideoFrame
import numpy as np
import threading
import asyncio
import queue
import mss

class ScreenCapturing(VideoStreamTrack):
    """
    自定义视频流轨道：实时捕获主显示器画面（monitor[1]）
    支持 WebRTC 兼容的 BGR24 格式帧输出与 PTS 时间戳管理。
    """
    def __init__(self) -> None:
        super().__init__()
        self.queue = queue.Queue(maxsize=10)  # 限制缓冲深度，防内存溢出
        self._started = False

    async def recv(self) -> VideoFrame:
        """核心方法：被 aiortc 调用以获取下一帧"""
        # 阻塞等待新帧（超时可选，避免永久挂起）
        try:
            img = self.queue.get(timeout=5.0)
        except queue.Empty:
            # 若长时间无帧，生成黑帧兜底（可选健壮性增强）
            dummy = np.zeros((720, 1280, 3), dtype=np.uint8)
            frame = VideoFrame.from_ndarray(dummy, format="bgr24")
            pts, time_base = await self.next_timestamp()
            frame.pts = pts
            frame.time_base = time_base
            return frame

        # RGBA → RGB（丢弃 Alpha 通道），确保格式兼容
        if img.shape[2] == 4:
            img_rgb = img[:, :, :3]
        else:
            img_rgb = img

        # 创建 VideoFrame 并设置时间戳（必需！否则播放卡顿/不同步）
        frame = VideoFrame.from_ndarray(img_rgb, format="bgr24")
        pts, time_base = await self.next_timestamp()
        frame.pts = pts
        frame.time_base = time_base
        return frame

    def start(self) -> None:
        """启动独立捕获线程（daemon=True 确保主程序退出时自动终止）"""
        if self._started:
            return
        self._started = True
        thread = threading.Thread(target=self._capture_loop, daemon=True)
        thread.start()

    def _capture_loop(self) -> None:
        """后台捕获循环：使用 mss 抓取主屏，转为 ndarray 后入队"""
        with mss.mss() as sct:
            monitor = sct.monitors[1]  # [0] 是虚拟全屏，[1] 是主显示器
            while self._started:
                try:
                    # 高效抓屏（返回 PIL Image，转 ndarray 开销可控）
                    im = sct.grab(monitor)
                    im_np = np.array(im)
                    # 尝试非阻塞入队，满则丢弃旧帧（保证低延迟）
                    try:
                        self.queue.put_nowait(im_np)
                    except queue.Full:
                        # 丢弃最老帧，优先保障实时性
                        try:
                            self.queue.get_nowait()
                            self.queue.put_nowait(im_np)
                        except queue.Empty:
                            pass
                except Exception as e:
                    print(f"[ScreenCapturing] 捕获异常: {e}")
                    break

    def stop(self) -> None:
        """停止捕获（配合 on_shutdown 使用）"""
        self._started = False
        # 清空队列避免残留引用
        while not self.queue.empty():
            try:
                self.queue.get_nowait()
            except queue.Empty:
                break

接下来，在主服务逻辑中替换原有的 create_local_tracks 函数，并确保在 offer 处理流程中正确初始化和启动该轨道：

# 替换原 create_local_tracks 函数
async def create_local_tracks():
    track = ScreenCapturing()
    track.start()  # ⚠️ 必须显式调用启动捕获线程
    return track

# 修改 offer 处理函数（关键变更点）
async def offer(request):
    params = await request.json()
    offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])
    pc = RTCPeerConnection()
    pcs.add(pc)

    # ✅ 使用实时屏幕轨道替代 MediaPlayer
    video = await create_local_tracks()  # 注意：此处为协程，需 await
    pc.addTrack(video)

    await pc.setRemoteDescription(offer)
    answer = await pc.createAnswer()
    await pc.setLocalDescription(answer)

    return web.Response(
        content_type="application/json",
        text=json.dumps(
            {"sdp": pc.localDescription.sdp, "type": pc.localDescription.type}
        ),
    )

重要注意事项与优化建议：

• 显示器选择：sct.monitors[1] 对应主显示器；可通过 print(sct.monitors) 查看所有屏幕区域，按需调整索引或指定 {"top": ..., "left": ..., "width": ..., "height": ...}。
• 性能调优：
  • 降低分辨率（如 monitor.update({"width": 1280, "height": 720})）可显著提升帧率；
  • mss 默认使用 numpy 后端，无需额外配置；
  • 避免在 recv() 中执行耗时操作（如图像缩放、编码），应在捕获线程中预处理。
• 时间戳（PTS）：await self.next_timestamp() 是强制要求——它确保帧按正确时间间隔渲染。忽略会导致浏览器解码器卡顿、音画不同步。
• 资源清理：在 on_shutdown 中，除关闭 RTCPeerConnection 外，建议调用 track.stop()（若已扩展该接口），防止后台线程残留。
• 安全性与权限：Windows/macOS/Linux 均需授予 Python 进程屏幕录制权限（如 macOS 的“屏幕录制”隐私设置），否则 mss.grab() 将返回黑屏或报错。

通过以上改造，你的 aiortc 服务即可从“播放录像”升级为“直播桌面”，真正实现 WebRTC 场景下的实时协作、远程支持与教学演示能力。整个方案不依赖 FFmpeg 或 GStreamer，轻量可靠，适合嵌入边缘设备或容器化部署。