背景

AI数字人直播的核心优势之一是"一套源流同时推流到多个平台"。但多平台推流不是简单地把同一个RTMP地址复制N份——不同平台对码率、分辨率、推流协议的要求不同，单平台断流不能影响其他平台。本文设计一套通用的多平台推流调度系统架构。

系统核心目标

• 同一AI渲染源流同时分发到N个平台（N≤20）
• 每路推流独立编码，适配各平台码率/分辨率要求
• 单平台断流自动重连，不影响其他平台
• 实时监控各平台推流状态，异常告警
• 支持动态增减推流目标（不停播增减平台）

系统架构

系统分四层：源流层、转码适配层、推流执行层、

监控调度层。

源流层

接收AI渲染引擎输出的原始音视频流：

# 伪代码：源流接入
class SourceStream:
    raw_video: VideoFrame   # YUV420p原始帧
    raw_audio: AudioFrame   # PCM原始音频
    fps: int                # 帧率（25fps）
    resolution: tuple       # 分辨率（1080x1920）

    def get_frame(self) -> tuple:
        """获取下一帧音视频数据"""
        return (self.raw_video, self.raw_audio)

转码适配层

核心设计：源流只渲染一次，转码层为每个平台生成独立的编码流。

# 伪代码：多平台转码适配
class TranscodeAdapter:
    def __init__(self, platform_configs: list):
        self.encoders = []
        for config in platform_configs:
            encoder = VideoEncoder(
                codec=config.codec,          # H.264/H.265
                bitrate=config.max_bitrate,   # 平台码率上限
                resolution=config.resolution, # 平台分辨率要求
                fps=config.fps                # 平台帧率要求
            )
            self.encoders.append(encoder)

    def process_frame(self, source_frame: VideoFrame) -> list:
        """一帧源流→N路编码流"""
        encoded_streams = []
        for encoder in self.encoders:
            # 分辨率适配（裁剪/缩放）
            adapted = self.adapt_resolution(source_frame, encoder.resolution)
            # 编码
            encoded = encoder.encode(adapted)
            encoded_streams.append(encoded)
        return encoded_streams

推流执行层

每路推流在独立线程/协程中运行，互不影响：

# 伪代码：推流执行器
class StreamPusher:
    def __init__(self, platform_config: dict):
        self.rtmp_url = platform_config["rtmp_url"]
        self.stream_key = platform_config["stream_key"]
        self.max_retries = 3
        self.retry_interval = 5  # 秒
        self.is_alive = True

    def push(self, encoded_stream: bytes):
        """持续推流，断线自动重连"""
        while self.is_alive:
            try:
                connection = self.connect(self.rtmp_url, self.stream_key)
                for data in encoded_stream:
                    connection.send(data)
            except ConnectionError:
                # 断线重连，不影响其他推流
                for attempt in range(self.max_retries):
                    time.sleep(self.retry_interval)
                    if self.reconnect():
                        break
                else:
                    self.notify_failure()

监控调度层

# 伪代码：推流状态监控
class PushMonitor:
    def __init__(self, pushers: list):
        self.pushers = pushers
        self.status_map = {}

    def health_check(self):
        """定时检查各平台推流健康状态"""
        for pusher in self.pushers:
            status = {
                "platform": pusher.platform_name,
                "is_alive": pusher.is_alive,
                "uptime": pusher.get_uptime(),
                "fps": pusher.get_actual_fps(),
                "bitrate": pusher.get_actual_bitrate(),
                "reconnect_count": pusher.reconnect_count,
            }
            self.status_map[pusher.platform_name] = status

            # 异常告警
            if status["fps"] < 20 or status["reconnect_count"] > 2:
                self.send_alert(status)

关键设计点

1. 源流单次渲染：AI渲染是最耗资源的环节（GPU），源流只渲染一次，转码层负责多路适配。不能为每个平台独立渲染。

2. 推流隔离：每个平台一个独立推流线程/协程，一个平台断流不影响其他平台。这是多平台推流的核心原则。

3. 分辨率适配策略：

平台类型	分辨率要求	适配方式
抖音/快手	1080×1920竖屏	源流竖屏直接用
淘宝/京东	1920×1080横屏	旋转+裁剪
视频号	1080×1920竖屏	源流竖屏直接用
小红书	1080×1920竖屏	源流竖屏直接用

1. 码率自适应：根据平台要求动态调整编码码率。淘宝上限8000kbps，抖音上限6000kbps，超出会被平台转码降质。

2. 动态增减平台：支持在直播过程中动态添加或移除推流目标。比如某个平台突然封号，可以立即移除该路推流，不影响其他平台。

常见坑

• 共享连接导致连锁故障：多个平台共用一个推流连接，一个断了全断。必须每平台独立连接。
• 转码资源不够：20个平台同时转码对CPU/GPU压力很大。建议用硬件编码器（NVENC）或限制同时推流平台数（10个以内）。
• 忽略平台推流地址过期：部分平台推流地址有时效性（如2小时过期），需要在地址过期前重新获取。
• 音频不同步：多路转码时视频和音频可能产生不同步偏差，需要在编码后做音视频对齐校验。

总结

多平台推流调度系统的核心设计原则是：源流单次渲染、转码多路适配、推流独立隔离、监控实时告警。技术上最大的挑战不是"能不能同时推流"，而是"一个平台出问题不影响其他平台"的故障隔离能力。建议系统上线前做72小时稳定性测试，覆盖断网重连、地址过期、码率超限等异常场景。