私有化部署 WebRTC：基于 aiortc 实现 Web 浏览器直接预览远程摄像头

三、详细部署步骤

步骤 1：环境准备

在开始之前，请确保你有一台具有公网 IP 的云服务器（如阿里云 ECS）。我们将在这台服务器上部署所有服务。

首先设置环境变量（这些变量将在后续脚本中使用）：

export ECS_IP_ADDR=替换为你的云服务器实际 IP 地址
export STUN_SRV=$ECS_IP_ADDR
export STUN_PORT=3478
export SIGNALING_SRV=$ECS_IP_ADDR  # 信令服务器，用来交换 WebRTC 连接需要的信息
export SIGNALING_PORT=9901
export WEB_SRV=$ECS_IP_ADDR  # Web 服务器
export WEB_PORT=9900

重要提示：将这些变量添加到你的 ~/.bashrc 文件中，以便每次登录时自动加载。

步骤 2：部署 STUN/TURN 服务器

STUN/TURN 服务器是 WebRTC 的"网络向导"，帮助两个设备找到彼此并建立连接。

2.1 安装 coturn

# 在 CentOS/RHEL 系统上安装
yum install coturn -y

# 配置 coturn
cat > /etc/coturn/turnserver.conf << EOF
listening-ip=0.0.0.0
listening-port=$STUN_PORT
relay-ip=$(ifconfig eth0 | grep -w inet | awk '{print $2}')
external-ip=$ECS_IP_ADDR
log-file=/var/tmp/turn.log
min-port=40000
max-port=65535
EOF

systemctl restart coturn
systemctl status coturn

2.2 配置防火墙

云服务器需要开放以下端口：

• 3478：STUN/TURN 服务端口
• 9900：Web 服务端口
• 9901：信令服务器端口
• 40000-65535：TURN 中继端口范围

在阿里云控制台的安全组中，添加入站规则允许这些端口的 TCP 和 UDP 流量。

2.3 测试 STUN/TURN 服务

访问 Trickle ICE

stun:你的服务器 IP:3478
turn:你的服务器 IP:3478

使用用户名 webrtc，密码 webrtc。如果测试成功，你会看到类似以下结果：

• srflx：STUN 穿透成功
• relay：TURN 中转可用

步骤 3：部署信令服务器

信令服务器是 WebRTC 连接的"媒人"，负责交换连接信息。它使用 WebSocket 协议进行实时通信。

3.1 创建信令服务器脚本

cat > signaling_server.py << 'EOF'
import asyncio
import json
import logging
from aiohttp import web
import aiohttp_cors

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SignalingServer:
    def __init__(self):
        self.app = web.Application()
        self.clients = {}  # 存储连接的用户
        self.setup_routes()

    def setup_routes(self):
        cors = aiohttp_cors.setup(self.app, defaults={
            "*": aiohttp_cors.ResourceOptions(
                allow_credentials=True,
                expose_headers="*",
                allow_headers="*",
            )
        })
        # 添加 WebSocket 路由
        resource = cors.add(self.app.router.add_resource('/ws'))
        cors.add(resource.add_route("GET", self.websocket_handler))
        # 健康检查路由
        self.app.router.add_get('/health', self.health_check)
        # 获取在线设备列表
        self.app.router.add_get('/devices', self.get_devices)

    async def health_check(self, request):
        return web.Response(text='OK')

    async def get_devices(self, request):
        """获取所有在线的设备"""
        devices = []
        for client_id, client_info in self.clients.items():
            if client_info['type'] == 'device':
                devices.append({
                    'id': client_id,
                    'type': client_info['type']
                })
        return web.json_response({'devices': devices})

    async def websocket_handler(self, request):
        ws = web.WebSocketResponse()
        await ws.prepare(request)
        client_id = None
        client_type = None
        try:
            async for msg in ws:
                if msg.type == web.WSMsgType.TEXT:
                    try:
                        data = json.loads(msg.data)
                        message_type = data.get('type')
                        if message_type == 'register':
                            # 客户端注册
                            client_id = data.get('client_id')
                            client_type = data.get('client_type')
                            # 'web' 或 'device'
                            if client_id in self.clients:
                                try:
                                    await self.clients[client_id]['ws'].close()
                                except:
                                    pass
                            self.clients[client_id] = {
                                'ws': ws,
                                'type': client_type,
                                'client_id': client_id
                            }
                            logger.info(f"Client registered: {client_id} ({client_type})")
                            await ws.send_str(json.dumps({'type': 'registered', 'status': 'success'}))
                        elif message_type == 'offer':
                            # 转发 offer 到对应的设备端
                            target_device = data.get('target_device')
                            offer_data = data.get('offer')
                            if target_device in self.clients:
                                await self.clients[target_device]['ws'].send_str(
                                    json.dumps({
                                        'type': 'offer',
                                        'offer': offer_data,
                                        'from_client': client_id
                                    })
                                )
                                logger.info(f"Offer forwarded from {client_id} to {target_device}")
                            else:
                                logger.warning(f"Target device not found: {target_device}")
                                await ws.send_str(json.dumps({'type': 'error', 'message': f'Device {target_device} not found'}))
                        elif message_type == 'answer':
                            # 转发 answer 到对应的网页端
                            target_web = data.get('target_web')
                            answer_data = data.get('answer')
                            if target_web in self.clients:
                                await self.clients[target_web]['ws'].send_str(
                                    json.dumps({
                                        'type': 'answer',
                                        'answer': answer_data,
                                        'from_client': client_id
                                    })
                                )
                                logger.info(f"Answer forwarded from {client_id} to {target_web}")
                        elif message_type == 'ice_candidate':
                            # 转发 ICE candidate
                            target = data.get('target')
                            candidate = data.get('candidate')
                            if target in self.clients:
                                await self.clients[target]['ws'].send_str(
                                    json.dumps({
                                        'type': 'ice_candidate',
                                        'candidate': candidate,
                                        'from_client': client_id
                                    })
                                )
                        elif message_type == 'ping':
                            # 心跳检测
                            await ws.send_str(json.dumps({'type': 'pong'}))
                    except json.JSONDecodeError as e:
                        logger.error(f"JSON 解析错误：{e}")
                elif msg.type == web.WSMsgType.ERROR:
                    logger.error(f'WebSocket error: {ws.exception()}')
        except Exception as e:
            logger.error(f"WebSocket error: {e}")
        finally:
            # 清理客户端
            if client_id and client_id in self.clients:
                del self.clients[client_id]
                logger.info(f"Client disconnected: {client_id}")
        return ws

def main():
    server = SignalingServer()
    logger.info(f"Starting signaling server on http://0.0.0.0:{os.environ.get('SIGNALING_PORT', 9901)}")
    web.run_app(server.app, host='0.0.0.0', port=int(os.environ.get('SIGNALING_PORT', 9901)))

if __name__ == '__main__':
    import os
    main()
EOF

3.2 安装依赖并运行

# 安装 Python 环境（如果使用 conda）
yum install conda -y
conda create -n myenv python=3.12
source ~/.bashrc
conda activate myenv

# 安装 Python 依赖
pip install aiohttp
pip install aiohttp_cors

# 注意：脚本中的环境变量需要替换为实际值
# 运行信令服务器
python signaling_server.py

步骤 4：部署 Web 服务器

Web 服务器提供用户访问的界面，包含 WebRTC JavaScript 客户端。

4.1 创建 Web 服务器

cat > web_server.py << 'EOF'
from aiohttp import web
import aiohttp_cors
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class WebServer:
    def __init__(self):
        self.app = web.Application()
        self.setup_routes()

    def setup_routes(self):
        cors = aiohttp_cors.setup(self.app, defaults={
            "*": aiohttp_cors.ResourceOptions(
                allow_credentials=True,
                expose_headers="*",
                allow_headers="*",
            )
        })
        # 主页面
        self.app.router.add_get('/', self.index_handler)
        self.app.router.add_get('/index.html', self.index_handler)
        # 设备端页面
        self.app.router.add_get
EOF