实时音视频 (LiveKit)

📋 模块概述

实时音视频通讯模块基于LiveKit SDK实现,提供WebRTC音视频实时通信功能。该模块支持低延迟音视频通话、实时水印叠加,广泛应用于远程监控、视频会议、现场指挥等场景。

核心类:

• android.znhaas.ui.MainActivity (LiveKit集成)
• android.znhaas.livekit.processor.CustomTextFrameProcessor (视频水印)
• android.znhaas.livekit.processor.CustomGpuTextFrameProcessor (GPU加速)
• android.znhaas.livekit.processor.CustomCapturePostProcessor (音频处理)

技术栈:

• LiveKit SDK: 开源WebRTC SFU解决方案
• Camera2 API: 高性能相机采集
• OpenGL ES: GPU加速视频处理
• WebRTC原生音频处理: AEC/NS/AGC

🎯 主要功能

• ✅ 低延迟实时音视频 (延迟小于500ms)
• ✅ 多方通话支持
• ✅ 自适应码率和分辨率
• ✅ 自定义视频水印(设备ID、GPS、时间戳)
• ✅ 音频回声消除(AEC)、噪声抑制(NS)
• ✅ 啸叫检测和抑制
• ✅ 硬件编解码加速(H.264/VP8)

📚 架构设计

1. LiveKit 房间连接流程

┌─────────────┐      ┌─────────────┐      ┌─────────────┐
│  MQTT指令   │─────▶│ connectRoom │─────▶│ LiveKit房间 │
│  (9527)     │      │   (token)   │      │  (WebRTC)   │
└─────────────┘      └─────────────┘      └─────────────┘
       │                    │                     │
       │                    ▼                     ▼
       │            ┌────────────────┐    ┌────────────────┐
       │            │  创建视频轨道   │    │  发布音频轨道   │
       │            │ (LocalVideo)   │    │ (LocalAudio)   │
       │            └────────────────┘    └────────────────┘
       │                    │                     │
       └────────────────────┼─────────────────────┘
                            ▼
                    ┌────────────────┐
                    │ 开始音视频传输 │
                    └────────────────┘

2. 视频处理管道

┌───────────┐    ┌──────────────────┐    ┌─────────────────┐
│  Camera2  │───▶│ VideoProcessor   │───▶│  WebRTC编码器   │
│  (采集)   │    │  (水印叠加)      │    │   (H.264)       │
└───────────┘    └──────────────────┘    └─────────────────┘
                         │                        │
                         ▼                        ▼
              ┌──────────────────┐      ┌─────────────────┐
              │ I420Buffer处理   │      │  LiveKit发布    │
              │ (YUV格式转换)    │      │  (推送到服务器)  │
              └──────────────────┘      └─────────────────┘

3. 音频处理管道

┌───────────┐    ┌──────────────────┐    ┌─────────────────┐
│ 麦克风采集 │───▶│ AudioProcessor   │───▶│  WebRTC编码器   │
│           │    │ (啸叫/回声抑制)   │    │   (Opus)        │
└───────────┘    └──────────────────┘    └─────────────────┘
                         │                        │
                         ▼                        ▼
              ┌──────────────────┐      ┌─────────────────┐
              │ 增益控制/噪声门   │      │  LiveKit发布    │
              │ (Gain/NoiseGate) │      │  (推送到服务器)  │
              └──────────────────┘      └─────────────────┘

📚 核心API

1. 初始化LiveKit房间

// MainActivity.kt

// 创建视频处理器
private var vedioProcessor: CustomTextFrameProcessor? = null

// 初始化LiveKit
private var room: Room? = null

// 初始化方法
private fun initLiveKit() {
    // 创建视频处理器
    vedioProcessor = CustomTextFrameProcessor().apply {
        setId(deviceId)  // 设置设备ID
        setLocationService(locationService)  // 设置定位服务
        setRotation(0)  // 设置旋转角度 (0, 90, 180, 270)
    }
    
    // 创建音频处理器
    val audioProcessor = CustomCapturePostProcessor()
    
    // 创建LiveKit房间
    room = LiveKit.create(
        appContext = applicationContext,
        overrides = LiveKitOverrides(
            audioOptions = AudioOptions(
                audioProcessorOptions = AudioProcessorOptions(
                    // 启用WebRTC原生麦克风处理
                    capturePostBypass = false,
                    capturePostProcessor = audioProcessor,  // 自定义音频处理
                    // 启用WebRTC原生扬声器处理
                    renderPreBypass = false,
                    renderPreProcessor = null
                )
            )
        ),
        options = RoomOptions()
    )
}

2. 连接房间并发布音视频

// 连接到LiveKit房间
suspend fun connectRoom(
    webrtcServer: String,  // WebRTC服务器地址
    token: String,         // JWT Token
    microphoneEnabled: Boolean = true,
    cameraEnabled: Boolean = true
) {
    try {
        // 1. 连接到房间
        room?.connect(webrtcServer, token)
        logUtil.d(TAG, "已连接到LiveKit房间")
        
        // 2. 获取本地参与者
        val localParticipant = room?.localParticipant
        
        // 3. 启用麦克风
        localParticipant?.setMicrophoneEnabled(microphoneEnabled)
        
        // 4. 创建视频轨道 (如果相机启用)
        if (cameraEnabled) {
            // 停止本地录制 (避免相机资源冲突)
            if (viewModel.recorderStatus.value == true) {
                recordingPresenter.stopRecording()
            }
            
            // 等待相机资源释放
            delay(200)
            
            // 创建视频轨道
            val localVideoTrack = localParticipant?.createVideoTrack(
                name = "camera",
                options = LocalVideoTrackOptions(
                    position = CameraPosition.BACK,  // 后置摄像头
                    captureParams = VideoCaptureParameter(
                        width = 1920,
                        height = 1080,
                        fps = 30
                    )
                ),
                videoProcessor = vedioProcessor  // 传入视频处理器
            )
            
            // 启动相机采集
            localVideoTrack?.startCapture()
            
            // 发布视频轨道
            localParticipant?.publishVideoTrack(localVideoTrack)
            
            logUtil.d(TAG, "视频轨道已发布")
        }
        
    } catch (e: Exception) {
        logUtil.e(TAG, "连接LiveKit失败: ${e.message}", e)
    }
}

3. 断开连接

fun disconnectRoom() {
    try {
        // 断开LiveKit连接
        room?.disconnect()
        logUtil.d(TAG, "已断开LiveKit连接")
        
        // 等待推流完全停止
        Thread.sleep(300)
        
        // 可选: 恢复本地录制
        if (recorderSwitcher) {
            Thread.sleep(500)  // 等待相机资源完全释放
            recordingPresenter.startRecording()
        }
        
    } catch (e: Exception) {
        logUtil.e(TAG, "断开LiveKit失败: ${e.message}", e)
    }
}

4. 释放资源

override fun onDestroy() {
    super.onDestroy()
    
    try {
        // 释放LiveKit资源
        room?.release()
        logUtil.d(TAG, "LiveKit资源已释放")
    } catch (e: Exception) {
        logUtil.e(TAG, "释放LiveKit失败: ${e.message}", e)
    }
}

📚 视频处理器

1. CustomTextFrameProcessor (CPU处理)

功能: 在视频帧上叠加文字水印

class CustomTextFrameProcessor : VideoProcessor {
    // 设置设备ID
    fun setId(id: String?) {
        mid = id
    }
    
    // 设置定位服务
    fun setLocationService(locationService: NativeLocationService?) {
        mlocationService = locationService
    }
    
    // 设置旋转角度
    fun setRotation(degrees: Int) {
        mrotationDegrees = degrees  // 0, 90, 180, 270
    }
    
    // 处理视频帧
    override fun onFrameCaptured(frame: VideoFrame?) {
        if (frame == null) return
        
        // 1. 转换为I420Buffer
        val i420 = frame.buffer.toI420()
        
        // 2. 转换为Bitmap
        val bitmap = i420ToArgbBitmap(i420, false)
        
        // 3. 叠加文字水印
        val canvas = Canvas(bitmap)
        val timeStr = timeFormat.format(Date())
        val locationText = getCurrentLocationText()  // 经度:纬度
        val osdText = "$mid  $locationText  $timeStr"
        
        canvas.drawText(osdText, 10f, 30f, overlayPaint)
        
        // 4. 转回I420Buffer
        val newI420 = bitmapToI420Buffer(bitmap)
        
        // 5. 创建新的VideoFrame (应用旋转)
        val processed = VideoFrame(newI420, mrotationDegrees, frame.timestampNs)
        
        // 6. 推送到下游
        sink?.onFrame(processed)
    }
}

使用示例:

val processor = CustomTextFrameProcessor().apply {
    setId("H07-001")
    setLocationService(locationService)
    setRotation(0)
}

// 在创建视频轨道时传入
val videoTrack = localParticipant?.createVideoTrack(
    name = "camera",
    options = LocalVideoTrackOptions(/*...*/),
    videoProcessor = processor
)

2. CustomGpuTextFrameProcessor (GPU加速)

功能: 使用OpenGL ES GPU加速文字叠加,性能更高

优势:

• 🚀 GPU加速,CPU占用低
• 💾 文字纹理缓存,避免重复渲染
• ⚡ 直接在Y平面叠加,无需全帧转换

class CustomGpuTextFrameProcessor : VideoProcessor {
    // 延迟初始化OpenGL (在第一帧处理时)
    override fun onFrameCaptured(frame: VideoFrame?) {
        if (frame == null) return
        
        if (!mOpenGLInitialized) {
            initOpenGL()  // 初始化OpenGL资源
            mOpenGLInitialized = true
        }
        
        val processed = processVideoFrameWithOpenGL(frame)
        sink?.onFrame(processed)
    }
    
    // GPU处理 - 直接在Y平面添加文字
    private fun createTextOverlayI420Buffer(originalI420: VideoFrame.I420Buffer): VideoFrame.I420Buffer {
        val width = originalI420.width
        val height = originalI420.height
        
        // 1. 复制原始I420数据
        val outputI420 = JavaI420Buffer.allocate(width, height)
        copyI420Buffer(originalI420, outputI420)
        
        // 2. 在Y平面添加文字水印 (使用缓存)
        addTextToYPlane(outputI420.dataY, outputI420.strideY, width, height)
        
        return outputI420
    }
}

📚 音频处理器

CustomCapturePostProcessor (啸叫抑制)

功能:

• 🔇 啸叫检测和实时抑制
• 🔄 回声消除
• 📊 噪声门控
• 🎤 麦克风增益控制

核心逻辑:

class CustomCapturePostProcessor : AudioProcessorInterface {
    // 音频处理流程
    private fun processLowSensitivityAudio(samples: ShortArray) {
        // 1. 麦克风增益控制 (降低灵敏度)
        applyMicrophoneGainControl(samples)
        
        // 2. 啸叫检测
        detectFeedback(samples)
        
        // 3. 啸叫抑制 (如果检测到)
        if (feedbackDetected) {
            applyFeedbackSuppression(samples)
        } else {
            // 4. 强噪声门处理
            applyStrongNoiseGate(samples)
            
            // 5. 回声抑制
            applyEchoSuppression(samples)
            
            // 6. 语音检测和优化
            detectVoice(samples)
        }
        
        // 7. 最终限幅
        applyFinalLimiting(samples)
    }
    
    // 啸叫检测
    private fun detectFeedback(samples: ShortArray) {
        var peakLevel = 0.0f
        
        for (sample in samples) {
            val absSample = abs(sample.toFloat()) / Short.MAX_VALUE
            if (absSample > peakLevel) {
                peakLevel = absSample
            }
        }
        
        // 啸叫特征: 高电平 + 稳定频率
        val isHighLevel = peakLevel > 0.1f
        val isRising = peakLevel > lastPeakLevel * 1.005f
        
        if (isHighLevel && isRising) {
            feedbackCounter++
            if (feedbackCounter >= feedbackThreshold) {
                feedbackDetected = true
                isMuteMode = true  // 触发静音模式
                android.util.Log.w(TAG, "检测到啸叫,启动抑制模式")
            }
        }
    }
    
    // 啸叫抑制 (完全静音)
    private fun applyFeedbackSuppression(samples: ShortArray) {
        if (isMuteMode) {
            for (i in samples.indices) {
                samples[i] = 0  // 完全静音
            }
        }
    }
}

📚 MQTT远程控制

开始推流指令

MQTT Topic: cmd/{deviceId}

指令格式 (JSON):

{
  "cmd": "9527",
  "webrtc_server": "wss://livekit.example.com",
  "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "mic_enable": true,
  "camera_enable": true,
  "resolution": "1920x1080"
}

字段说明:

• cmd: 固定为 "9527" (开始推流)
• webrtc_server: LiveKit服务器地址
• token: JWT认证令牌
• mic_enable: 是否启用麦克风
• camera_enable: 是否启用相机
• resolution: 视频分辨率 (1920x1080, 1280x720, 640x480)

停止推流指令

{
  "cmd": "9528"
}

处理逻辑:

// MainActivity.kt - MQTT消息回调
override fun messageArrived(topic: String?, message: MqttMessage?) {
    val payload = message?.toString() ?: return
    val webRtcMessage = Json.decodeFromString<WebRtcMessage>(payload)
    
    when (webRtcMessage.cmd) {
        START_WEBRTC -> {  // "9527"
            // 开始推流
            lifecycleScope.launch {
                connectRoom(
                    webrtcServer = webRtcMessage.webrtc_server,
                    token = webRtcMessage.token,
                    microphoneEnabled = webRtcMessage.mic_enable ?: true,
                    cameraEnabled = webRtcMessage.camera_enable ?: true
                )
            }
        }
        
        STOP_WEBRTC -> {  // "9528"
            // 停止推流
            disconnectRoom()
        }
    }
}

📚 分辨率配置

支持的分辨率

private fun switchLocalVideoCapture(resolution: String): VideoCaptureParameter {
    return when (resolution) {
        "1920x1080" -> VideoCaptureParameter(1920, 1080, 30)
        "1280x720"  -> VideoCaptureParameter(1280, 720, 30)
        "640x480"   -> VideoCaptureParameter(640, 480, 30)
        else        -> VideoCaptureParameter(1280, 720, 30)  // 默认720p
    }
}

动态切换分辨率

// 在推流过程中切换分辨率
val newCaptureParams = switchLocalVideoCapture("1920x1080")
localVideoTrack?.switchCamera(newCaptureParams)

🔧 注意事项

1. 相机资源冲突

问题: 本地录制和LiveKit推流不能同时使用相机

解决方案:

// 开始推流前,先停止本地录制
if (viewModel.recorderStatus.value == true) {
    recordingPresenter.stopRecording()
}

// 等待相机资源释放
delay(200)

// 然后创建LiveKit视频轨道
val videoTrack = localParticipant?.createVideoTrack(/*...*/)

2. Camera2异常处理

问题: 某些设备的Camera2 API存在兼容性问题

解决方案:

try {
    room?.connect(webrtcServer, token)
} catch (e: Exception) {
    if (e.message?.contains("Camera2", ignoreCase = true) == true ||
        e.message?.contains("stopRepeating", ignoreCase = true) == true) {
        logUtil.w(TAG, "检测到Camera2异常,这是已知的硬件兼容性问题")
        // 继续执行,不影响音频推流
    } else {
        throw e  // 重新抛出其他异常
    }
}

3. 权限要求

<!-- AndroidManifest.xml -->
<uses-permission android:name="android.permission.CAMERA" />
<uses-permission android:name="android.permission.RECORD_AUDIO" />
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />

4. 网络要求

• ✅ 推荐: WiFi或4G/5G网络
• ✅ 最低上行带宽: 1Mbps (720p), 2Mbps (1080p)
• ✅ 延迟: 局域网小于100ms, 公网小于500ms

5. 性能优化

建议:

• 使用CustomGpuTextFrameProcessor而非CustomTextFrameProcessor (GPU加速)
• 根据网络状况动态调整分辨率
• 在弱网环境下降低帧率 (15fps)
• 启用硬件编解码

📖 相关资源

源码位置

• MainActivity: app/src/main/java/android/znhaas/ui/MainActivity.kt
• 视频处理器: app/src/main/java/android/znhaas/livekit/processor/
• 音频处理器: app/src/main/java/android/znhaas/livekit/processor/CustomCapturePostProcessor.kt

依赖库

// app/build.gradle.kts
dependencies {
    // LiveKit SDK
    implementation("io.livekit:livekit-android:1.x.x")
    
    // WebRTC (自动依赖)
    // implementation("io.livekit:webrtc-android:xxx")
}

扩展阅读

• LiveKit官方文档
• WebRTC标准
• Camera2 API文档
• OpenGL ES教程

---

最后更新: 2025-01-19
文档版本: v1.0
基于代码版本: H07 Android App (main分支)