Python-视频直播(LIVE)-阿里云帮助中心

本文介绍如何在 Linux Python 项目中集成 ARTC SDK，快速实现一个简单的实时音视频互动程序，适用于视频会议、互动直播、云端录制等服务端场景。

功能简介

在开始之前，了解以下几个关键概念会很有帮助：

ARTC SDK：这是阿里云的实时音视频产品，帮助开发者快速实现实时音视频互动的SDK。
GRTN：阿里云全球实时传输网络，提供超低延时、高音质、安全可靠的音视频通讯服务。
频道：相当于一个虚拟的房间，所有加入同一频道的用户都可以进行实时音视频互动。
主播：可在频道内发布音视频流，并可订阅其他主播发布的音视频流。
观众：可在频道内订阅音视频流，不能发布音视频流。

实现实时音视频互动的基本流程如下：

用户需要调用setChannelProfile（设置频道场景），后调用joinChannel加入频道：
- 视频通话场景：所有用户都是主播角色，可以进行推流和拉流。
- 互动直播场景：需要调用setClientRole（设置角色），在频道内推流的用户设置主播角色；如果用户只需要拉流，不需要推流，则设置观众角色。
加入频道后，不同角色的用户有不同的推拉流行为：
- 所有加入频道内的用户都可以接收频道内的音视频流。
- 主播角色可以在频道内推音视频流。
- 观众如果需要推流，需要调用setClientRole方法，将用户角色切换成主播，便可以推流。

示例项目

SDK 包中提供了示例程序：

示例	文件	说明
功能演示	`demo.py`	包含完整推拉流、Token 生成等功能演示

运行示例：

cd Python
python3 demo.py

前置条件

操作系统：Linux（内核 2.6+）。
Python 版本：Python 3.6+。
网络环境：稳定的互联网连接。
应用准备：获取实时音视频应用的 AppID 和 AppKey，详情请参见创建应用。

实现步骤

步骤一：导入 SDK

SDK 包目录结构如下：

AliRTCSDK_Linux/
└── Python/
    ├── Release/
    │   └── lib/
    │       ├── AliRtcCoreService             # 后台服务进程（须指定绝对路径）
    │       ├── libAliRtcLinuxEngine.so        # SDK 完整动态库
    │       └── libonnxruntime.so.1.16.3       # AI 降噪依赖库
    ├── AliRTCEngine.py                        # Python 接口封装
    ├── AliRTCEngineImpl.py                    # 接口实现
    ├── AliRTCLinuxSdkDefine.py                # 数据结构与枚举定义
    └── demo.py                                # 功能演示示例

在您的 Python 文件中导入 SDK：

from AliRTCLinuxSdkDefine import *
import AliRTCEngine

运行前设置动态库搜索路径：

export LD_LIBRARY_PATH=/path/to/Python/Release/lib:$LD_LIBRARY_PATH

步骤二：实现事件回调类

继承 AliRTCEngine.EngineEventHandlerInterface 实现事件回调，用于接收 SDK 推送的各类通知。

import AliRTCEngine
from AliRTCLinuxSdkDefine import ERROR_CODE

class VideoCallEventHandler(AliRTCEngine.EngineEventHandlerInterface):

    def OnJoinChannelResult(self, result: int, channel: str, userId: str) -> None:
        if result == 0:
            print(f"[OnJoinChannelResult] User {userId} joined channel {channel} successfully")
        else:
            print(f"[OnJoinChannelResult] Failed to join, error: {result}")

    def OnRemoteUserOnLineNotify(self, uid: str) -> None:
        print(f"[OnRemoteUserOnLineNotify] uid: {uid}")

    def OnRemoteUserOffLineNotify(self, uid: str) -> None:
        print(f"[OnRemoteUserOffLineNotify] uid: {uid}")

    # OnSubscribeMixAudioFrame: 接收混音后的远端 PCM 音频帧
    # 订阅配置中 subscribeAudioFormat = AudioFormatMixedPcm 时触发
    def OnSubscribeMixAudioFrame(self, frame: AliRTCEngine.AudioFrame) -> None:
        # frame.data       PCM 数据（bytes，int16_t 格式）
        # frame.channels   声道数
        # frame.sampleRate 采样率
        # 在此写入文件、送入音频设备或解码播放
        pass

    # OnSubscribeAudioFrame: 接收逐路远端用户的未混音 PCM 帧，uid 区分不同远端用户的音频流
    # 订阅配置中 subscribeAudioFormat = AudioFormatPcmBeforMixing 时触发
    def OnSubscribeAudioFrame(self, uid: str, frame: AliRTCEngine.AudioFrame) -> None:
        # uid 标识该帧来自哪个远端用户
        # 在此按用户分别处理音频数据
        pass

    # OnRemoteVideoSample: 接收远端视频帧
    # uid 区分不同远端用户的视频流
    def OnRemoteVideoSample(self, uid: str, frame: AliRTCEngine.VideoFrame) -> None:
        # uid 标识该帧来自哪个远端用户
        # 在此写入文件、送入渲染器或视频解码器
        pass

    def OnError(self, error_code: ERROR_CODE) -> None:
        print(f"[OnError] error_code: {error_code.value:#010x}")

步骤三：鉴权 Token

Python SDK 内置 GenerateToken 方法，可在客户端直接生成单参数入会 Token。

Token 生成流程：

拼接字符串：appId + appKey + channelId + userId + nonce + timestamp
SHA-256 哈希得到十六进制字符串
组装 JSON：{"appid":..., "channelid":..., "userid":..., "nonce":..., "timestamp":..., "token":<sha256>}
Base64 编码

import datetime
import time
from AliRTCLinuxSdkDefine import AuthInfo

authInfo = AuthInfo()
authInfo.appid    = "your_app_id"
authInfo.channel  = "your_channel_id"
authInfo.userid   = "your_user_id"
authInfo.username = "your_user_id"

expire = datetime.datetime.now() + datetime.timedelta(days=1)
authInfo.timestamp = int(time.mktime(expire.timetuple()))  # 24 小时后过期

app_key = "your_app_key"  # 生产环境请勿将 AppKey 暴露在客户端代码中

# 调用 GenerateToken 生成单参数入会 Token（需先创建引擎实例）
authInfo.token = linuxEngine.GenerateToken(authInfo, app_key)

说明

Token 安全提示：示例中在客户端本地生成 Token，仅适用于开发和测试阶段。生产环境中，Token 必须由您的业务服务端生成并下发，避免将 AppKey 暴露在客户端代码中。

步骤四：创建并初始化音视频引擎

调用 AliRTCEngine.CreateAliRTCEngine 创建引擎实例，并传入事件回调对象。

import json
import os
import AliRTCEngine

event_handler = VideoCallEventHandler()

current_path = os.getcwd()
core_service_path = os.path.abspath(
    os.path.join(current_path, "Release", "lib", "AliRtcCoreService")
)

extra_jobj = {"user_specified_disable_audio_ranking": "true"}
extra = json.dumps(extra_jobj)

h5mode = False  # 与 Web 端互通请设置为 True

linuxEngine = AliRTCEngine.CreateAliRTCEngine(
    event_handler,
    42000, 45000,       # IPC 端口范围，用于与 AliRtcCoreService 进程通信
    "/tmp",             # 日志文件目录
    core_service_path,  # AliRtcCoreService 绝对路径
    h5mode,
    extra
)

if linuxEngine is None:
    print("Failed to create RTC engine")
    exit(1)

说明

每创建一个引擎实例，SDK 会启动一个对应的 AliRtcCoreService 后台进程（对应一个虚拟用户）。

步骤五：设置音视频属性

调用 SetClientRole 设置用户角色，调用 SetVideoEncoderConfiguration 配置视频编码参数。

from AliRTCLinuxSdkDefine import (
    AliEngineClientRole, AliEngineVideoEncoderConfiguration,
    AliEngineFrameRate, AliEngineVideoEncoderOrientationMode,
    AliEngineVideoMirrorMode, AliEngineRotationMode
)

# 调用 SetClientRole 设置用户角色为互动模式（主播），可同时发布和订阅
linuxEngine.SetClientRole(AliEngineClientRole.AliEngineClientRoleInteractive)

# 调用 SetVideoEncoderConfiguration 设置视频编码参数
video_config = AliEngineVideoEncoderConfiguration(
    width=720, height=1280,
    f=AliEngineFrameRate.AliEngineFrameRateFps15,
    b=1200,
    ori=AliEngineVideoEncoderOrientationMode.AliEngineVideoEncoderOrientationModeAdaptive,
    mr=AliEngineVideoMirrorMode.AliEngineVideoMirrorModeDisabled,
    rotation=AliEngineRotationMode.AliEngineRotationMode_0
)
linuxEngine.SetVideoEncoderConfiguration(video_config)

步骤六：设置推拉流属性

配置音视频的发布和订阅行为，并启用 Linux 平台特有的外部音视频源模式。

from AliRTCLinuxSdkDefine import (
    VideoSource, RenderMode, JoinChannelConfig, ChannelProfile,
    PublishMode, SubscribeMode, PublishAvsyncMode, AudioFormat, VideoFormat
)

# 调用 PublishLocalVideoStream / PublishLocalAudioStream 开启本地音视频发布
linuxEngine.PublishLocalVideoStream(True)
linuxEngine.PublishLocalAudioStream(True)

# Linux 无内置摄像头/麦克风，调用 SetExternalVideoSource 启用外部视频源，
# 通过 PushExternalVideoFrame 输入 YUV 帧数据
linuxEngine.SetExternalVideoSource(True,
    sourceType=VideoSource.VideoSourceCamera,
    renderMode=RenderMode.RenderModeFill)

# 调用 SetExternalAudioSource 启用外部音频源，通过 PushExternalAudioFrameRawData 输入 PCM 帧数据
linuxEngine.SetExternalAudioSource(True, sampleRate=16000, channelsPerFrame=1)

配置入会时的订阅模式（在步骤七的 JoinChannelConfig 中设置）：

join_config = JoinChannelConfig()
join_config.channelProfile    = ChannelProfile.ChannelProfileInteractiveLive
join_config.publishMode       = PublishMode.PublishAutomatically      # 自动推流
join_config.subscribeMode     = SubscribeMode.SubscribeAutomatically  # 自动订阅
join_config.publishAvsyncMode = PublishAvsyncMode.PublishAvsyncWithPts

# 音频订阅格式有两种选择：
# AudioFormatMixedPcm: 接收混音后的整频道 PCM，触发 OnSubscribeMixAudioFrame
# AudioFormatPcmBeforMixing: 接收每个远端用户的逐路 PCM，触发 OnSubscribeAudioFrame（含 uid）
join_config.subscribeAudioFormat = AudioFormat.AudioFormatMixedPcm

# 接收 H264 视频帧，触发 OnRemoteVideoSample
join_config.subscribeVideoFormat = VideoFormat.VideoFormatH264

步骤七：加入频道

调用 JoinChannel 传入步骤三生成的 Token 和频道配置加入频道。

# 调用 JoinChannel 加入频道（单参数入会）
linuxEngine.JoinChannel(
    authInfo.token,
    authInfo.channel,
    authInfo.userid,
    authInfo.username,
    join_config
)

说明

请勿重复调用 JoinChannel。GenerateToken 接口仅供开发和测试使用，生产环境中请通过业务服务端获取 Token，避免 AppKey 泄漏。

步骤八：推送外部视频帧

Linux 平台没有内置摄像头驱动接口，通过外部输入将 YUV 视频数据送入 SDK。以下示例从 I420 格式 YUV 文件循环读取帧数据推送，实际业务中可替换为摄像头驱动或视频解码器输出。

import threading
import time
from AliRTCLinuxSdkDefine import VideoDataSample, VideoDataFormat, VideoBufferType, VideoSource

def push_video():
    width, height, fps = 720, 1280, 15
    frame_size = width * height * 3 // 2  # I420

    with open("/tmp/test_720p.yuv", "rb") as video_file:
        v_ts = 0
        while running:
            data = video_file.read(frame_size)
            if not data or len(data) < frame_size:
                video_file.seek(0)  # 文件读完后循环重读
                continue

            sample = VideoDataSample()
            sample.width      = width
            sample.height     = height
            sample.strideY    = width
            sample.strideU    = width // 2
            sample.strideV    = width // 2
            sample.dataLen    = frame_size
            sample.format     = VideoDataFormat.VideoDataFormatI420
            sample.bufferType = VideoBufferType.VideoBufferTypeRawData
            sample.rotation   = 0
            sample.data       = data
            sample.timeStamp  = v_ts

            linuxEngine.PushExternalVideoFrame(sample, VideoSource.VideoSourceCamera)
            v_ts += 1000 // fps
            time.sleep(1 / fps)

video_thread = threading.Thread(target=push_video)
video_thread.start()

步骤九：推送外部音频帧

Linux 平台没有内置麦克风录音接口，通过外部输入将 PCM 音频数据送入 SDK。以下示例从 PCM 文件（int16_t，16kHz，单声道）循环读取帧数据推送，实际业务中可替换为麦克风驱动或音频解码器输出。

def push_audio():
    sample_rate = 16000
    channels    = 1
    frame_ms    = 20  # 每帧 20ms
    frame_size  = (sample_rate // 1000) * frame_ms * 2 * channels  # int16_t = 2 bytes
    a_ts = 0

    with open("/tmp/test_16k_mono.pcm", "rb") as audio_file:
        while running:
            data = audio_file.read(frame_size)
            if not data or len(data) < frame_size:
                audio_file.seek(0)  # 文件读完后循环重读
                continue

            ret = linuxEngine.PushExternalAudioFrameRawData(data, frame_size, a_ts)
            if ret != 0:
                # SDK buffer 已满，回退文件指针并稍后重试
                audio_file.seek(audio_file.tell() - frame_size)
                time.sleep(0.02)
                continue

            a_ts += frame_ms
            time.sleep(frame_ms / 1000)

audio_thread = threading.Thread(target=push_audio)
audio_thread.start()

步骤十：处理远端音视频播放

Linux 平台没有内置音视频播放设备，远端音视频数据通过回调帧的方式交给应用层自行处理，例如写入文件、送入解码器或对接播放设备。

音频播放

根据步骤六中 subscribeAudioFormat 的配置，收到远端音频帧时会触发以下回调之一：

# AudioFormatMixedPcm 模式：接收所有远端用户混音后的整频道 PCM 数据
def OnSubscribeMixAudioFrame(self, frame: AliRTCEngine.AudioFrame) -> None:
    # frame.data       PCM 数据（bytes，int16_t 格式）
    # frame.channels   声道数
    # frame.sampleRate 采样率
    # 在此写入文件、送入音频设备或解码播放
    pass

# AudioFormatPcmBeforMixing 模式：按用户接收未混音的逐路 PCM 数据
def OnSubscribeAudioFrame(self, uid: str, frame: AliRTCEngine.AudioFrame) -> None:
    # uid 标识该帧来自哪个远端用户
    # 在此按用户分别处理音频数据
    pass

视频播放

收到远端视频帧时触发 OnRemoteVideoSample 回调，帧格式由步骤六中 subscribeVideoFormat 决定：

def OnRemoteVideoSample(self, uid: str, frame: AliRTCEngine.VideoFrame) -> None:
    # uid 标识该帧来自哪个远端用户
    # 在此写入文件、送入渲染器或视频解码器
    pass

步骤十一：离开频道并销毁引擎

正确释放资源，依次停止推流、离会、销毁引擎。

# 停止外部推流线程
running = False
video_thread.join()
audio_thread.join()

# 调用 PublishLocalVideoStream(False) / PublishLocalAudioStream(False) 停止发布
linuxEngine.PublishLocalVideoStream(False)
linuxEngine.PublishLocalAudioStream(False)

# 调用 LeaveChannel 离开频道
linuxEngine.LeaveChannel()

# 等待 OnLeaveChannelResult 回调（stop_signal 置 True）后再销毁引擎
while not stop_signal:
    time.sleep(1)

# 调用 Release 销毁引擎（须在 LeaveChannel 之后调用）
linuxEngine.Release()
linuxEngine = None

常见问题

Q：h5mode 什么时候设置为 True？

只有与 Web 端（H5 页面）互通时才需要开启。纯 Linux 端互通设置为 False 即可。

Q：Token 过期后如何处理？

监听 OnAuthInfoWillExpire 回调（Token 即将过期），重新生成 Token 并调用引擎刷新接口更新凭证，无需重新入会。

监听 OnAuthInfoExpired 回调（Token 已过期），需离开频道并以新 Token 重新入会。

Q：运行时提示找不到动态库

error while loading shared libraries: libAliRtcLinuxEngine.so: cannot open shared object file

解决：执行 export LD_LIBRARY_PATH=/path/to/Python/Release/lib:$LD_LIBRARY_PATH。