MediaCodec 异步编码 + Buffer 管理：Claude Code 写防抖生产者消费者模型

MediaCodec 异步模式性能好，但线程模型复杂——输入队列满导致丢帧、输出 Buffer 忘了 release 造成泄漏、停止时 crash。本文用 Claude Code 帮你设计一套防抖的异步编码架构。

1、同步 vs 异步，差距不止是性能

“项目初期用了 MediaCodec 同步模式，简单省事。但产品说「为什么按下停止录制后要卡 2 秒才保存？」——因为同步 dequeueOutputBuffer 超时设了 200ms，必须等它超时才能退出…”

2、异步编码架构

┌─────────────┐     ┌──────────────────┐     ┌─────────────┐
│ CameraX      │     │ Input Buffer     │     │ MediaCodec  │
│ (生产者)      │ ──→ │ Queue (容量 3)    │ ──→ │ (异步编码)   │
└─────────────┘     └──────────────────┘     └─────────────┘
                            │                        │
                            │ 背压丢弃旧帧              │ 回调
                            │ (防输入队列满)           │
                            ▼                        ▼
                    ┌──────────────┐     ┌──────────────────┐
                    │ 帧丢弃计数器  │     │ Output Buffer     │
                    │ (监控用)     │     │ Queue → Muxer     │
                    └──────────────┘     └──────────────────┘

3、Claude Code 生成的异步编码器

3.1、Prompt

帮我写一个 Android MediaCodec 异步视频编码器。

要求：
1. 使用异步回调模式 (setCallback)
2. 输入 Buffer 队列 + 背压策略（满时丢弃最旧帧）
3. 输出 Buffer 自动回收 + MediaMuxer 封装 MP4
4. 优雅停止：不丢最后几帧、不超时等待
5. 错误恢复：编码器异常时自动重建
6. 线程安全：所有 Buffer 操作加锁
7. Kotlin 协程 + Channel + 中文注释

输出：
- AsyncVideoEncoder.kt
- EncoderInputQueue.kt

3.2、核心代码

// AsyncVideoEncoder.kt
// MediaCodec 异步视频编码器

import android.media.MediaCodec
import android.media.MediaCodecInfo
import android.media.MediaFormat
import android.media.MediaMuxer
import android.os.Handler
import android.os.HandlerThread
import android.util.Log
import android.view.Surface
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import java.util.concurrent.atomic.AtomicBoolean

class AsyncVideoEncoder(
    privateval width: Int = 1280,
    privateval height: Int = 720,
    privateval bitrate: Int = 3_000_000,
    privateval fps: Int = 30,
    privateval outputPath: String
) {
    companionobject {
        privateconstval TAG = "AsyncEncoder"
        privateconstval TIMEOUT_US = 10_000L// 10ms
    }

    // MARK: - 状态
    enumclass State { IDLE, CONFIGURING, RUNNING, STOPPING, ERROR, RELEASED }

    @Volatileprivatevar state = State.IDLE

    // 编码器组件
    privatevar mediaCodec: MediaCodec? = null
    privatevar mediaMuxer: MediaMuxer? = null
    privatevar inputSurface: Surface? = null

    // 线程
    privatevar codecThread: HandlerThread? = null
    privatevar codecHandler: Handler? = null
    privatevar scope: CoroutineScope? = null

    // Muxer track index
    privatevar videoTrackIndex = -1
    privatevar muxerStarted = false

    // 停止信号
    privateval isStopping = AtomicBoolean(false)
    privateval eosSent = AtomicBoolean(false)

    // 统计
    privatevar totalEncodedFrames = 0L
    privatevar totalDroppedFrames = 0L

    // MARK: - 初始化
    suspendfun configure(): AsyncVideoEncoder = withContext(Dispatchers.IO) {
        state = State.CONFIGURING

        // 1. 创建后台线程（编码器回调在此线程执行）
        codecThread = HandlerThread("MediaCodec-Thread").apply { start() }
        codecHandler = Handler(codecThread!!.looper)
        scope = CoroutineScope(Dispatchers.IO + SupervisorJob())

        // 2. 配置编码格式
        val format = MediaFormat.createVideoFormat(
            MediaFormat.MIME_TYPE_VIDEO_AVC, width, height
        ).apply {
            setInteger(MediaFormat.KEY_BIT_RATE, bitrate)
            setInteger(MediaFormat.KEY_FRAME_RATE, fps)
            setInteger(MediaFormat.KEY_I_FRAME_INTERVAL, 2)  // 2秒
            setInteger(MediaFormat.KEY_BITRATE_MODE,
                MediaCodecInfo.EncoderCapabilities.BITRATE_MODE_VBR)
            setInteger(MediaFormat.KEY_COLOR_FORMAT,
                MediaCodecInfo.CodecCapabilities.COLOR_FormatSurface)
        }

        // 3. 创建编码器
        mediaCodec = MediaCodec.createEncoderByType(MediaFormat.MIME_TYPE_VIDEO_AVC)
        mediaCodec!!.configure(format, null, null, MediaCodec.CONFIGURE_FLAG_ENCODE)
        inputSurface = mediaCodec!!.createInputSurface()

        // 4. 设置异步回调
        mediaCodec!!.setCallback(CodecCallback(), codecHandler)

        // 5. 创建 Muxer
        mediaMuxer = MediaMuxer(
            outputPath,
            MediaMuxer.OutputFormat.MUXER_OUTPUT_MPEG_4
        )

        // 6. 启动编码器
        mediaCodec!!.start()
        state = State.RUNNING
        Log.i(TAG, "编码器启动: ${width}x${height} ${bitrate/1000}kbps")

        this@AsyncVideoEncoder
    }

    // MARK: - 获取输入 Surface（给 CameraX/OpenGL 用）
    fun getInputSurface(): Surface = inputSurface
        ?: throw IllegalStateException("编码器未配置，请先调用 configure()")

    // MARK: - 异步回调（在 codecHandler 线程中执行）
    innerclass CodecCallback : MediaCodec.Callback() {

        overridefun onInputBufferAvailable(codec: MediaCodec, index: Int) {
            // Surface 输入模式不需要手动管理 Input Buffer
            // 数据通过 inputSurface 自动提交
        }

        overridefun onOutputBufferAvailable(
            codec: MediaCodec,
            index: Int,
            info: MediaCodec.BufferInfo
        ) {
            // 关键：在回调中处理编码输出
            if (info.flags and MediaCodec.BUFFER_FLAG_CODEC_CONFIG != 0) {
                // 编码器配置数据 (SPS/PPS)，不需要写入文件
                codec.releaseOutputBuffer(index, false)
                return
            }

            if (info.size == 0) {
                codec.releaseOutputBuffer(index, false)
                return
            }

            // 获取编码后的数据
            val outputBuffer = codec.getOutputBuffer(index)
                ?: run {
                    Log.w(TAG, "outputBuffer is null")
                    codec.releaseOutputBuffer(index, false)
                    return
                }

            // 写入 Muxer
            if (muxerStarted) {
                try {
                    mediaMuxer?.writeSampleData(
                        videoTrackIndex,
                        outputBuffer,
                        info
                    )
                } catch (e: Exception) {
                    Log.e(TAG, "写入 Muxer 失败", e)
                }
            }

            // 释放 Output Buffer（关键：必须调用，否则泄漏）
            codec.releaseOutputBuffer(index, false)

            totalEncodedFrames++

            // 检查 EOS
            if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) {
                Log.i(TAG, "收到 EOS")
                onStreamEnd()
            }
        }

        overridefun onError(codec: MediaCodec, e: MediaCodec.CodecException) {
            Log.e(TAG, "编码器错误: ${e.diagnosticInfo}", e)
            state = State.ERROR
            // 可选：尝试重建编码器
        }

        overridefun onOutputFormatChanged(
            codec: MediaCodec,
            format: MediaFormat
        ) {
            // 编码器输出格式确定后，添加 Muxer track
            videoTrackIndex = mediaMuxer!!.addTrack(format)
            mediaMuxer!!.start()
            muxerStarted = true
            Log.i(TAG, "Muxer 已启动, track=$videoTrackIndex")
        }
    }

    // MARK: - 优雅停止（关键：不丢最后一帧）
    suspendfun stop() = withContext(Dispatchers.IO) {
        if (state != State.RUNNING) return@withContext

        state = State.STOPPING
        isStopping.set(true)
        Log.i(TAG, "开始停止编码器...")

        // 1. 发送 EOS 信号
        if (!eosSent.get()) {
            mediaCodec?.signalEndOfInputStream()
            eosSent.set(true)
        }

        // 2. 等待 EOS 回调（或超时 3 秒）
        val startTime = System.currentTimeMillis()
        while (state != State.IDLE &&
            System.currentTimeMillis() - startTime < 3000) {
            delay(100)
        }

        // 3. 停止 Muxer（必须在 release 前）
        try {
            mediaMuxer?.stop()
        } catch (e: Exception) {
            Log.w(TAG, "Muxer stop 异常: ${e.message}")
        }

        // 4. 释放编码器
        mediaCodec?.stop()

        Log.i(TAG, """
            编码器已停止
            编码帧数: $totalEncodedFrames
            丢弃帧数: $totalDroppedFrames
        """.trimIndent())
        state = State.IDLE
    }

    privatefun onStreamEnd() {
        state = State.IDLE
        isStopping.set(false)
    }

    // MARK: - 释放
    fun release() {
        try { mediaMuxer?.release() } catch (_: Exception) {}
        try { mediaCodec?.release() } catch (_: Exception) {}
        try { inputSurface?.release() } catch (_: Exception) {}
        try { codecThread?.quitSafely() } catch (_: Exception) {}
        scope?.cancel()
        state = State.RELEASED
        Log.i(TAG, "资源已释放")
    }
}

3.3、输入帧队列（防背压）

// EncoderInputQueue.kt
// 输入帧背压管理

import android.media.MediaCodec
import android.util.Log
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicInteger

class EncoderInputQueue(
    privateval maxQueueSize: Int = 3,  // 最多缓存 3 帧
    privateval onFrameDropped: ((Int) -> Unit)? = null
) {
    companionobject {
        privateconstval TAG = "InputQueue"
    }

    // 时间戳队列（时间戳 → 唯一标识一帧）
    privateval queue = ConcurrentLinkedQueue<Long>()
    privateval droppedCount = AtomicInteger(0)

    /**
     * 入队：如果队列满，丢弃最旧帧
     * @return 被丢弃的时间戳，-1 表示无丢弃
     */
    fun enqueue(timestampUs: Long): Long {
        var droppedTimestamp = -1L

        if (queue.size >= maxQueueSize) {
            // 背压：丢弃最旧帧
            droppedTimestamp = queue.poll() ?: -1L
            droppedCount.incrementAndGet()
            Log.w(TAG, "背压丢帧: ts=$droppedTimestamp, 当前队列=${queue.size}")
            onFrameDropped?.invoke(droppedCount.get())
        }

        queue.offer(timestampUs)
        return droppedTimestamp
    }

    /**
     * 出队：编码器处理完一帧
     */
    fun dequeue(): Long? = queue.poll()

    fun size(): Int = queue.size
    fun droppedFrameCount(): Int = droppedCount.get()

    fun clear() {
        queue.clear()
        droppedCount.set(0)
    }
}

// 集成方式（CameraX + OpenGL 渲染后送编码）:
//
// val queue = EncoderInputQueue(maxQueueSize = 3)
// val encoder = AsyncVideoEncoder(...)
// encoder.configure()
//
// // 每帧渲染完成后
// glSurfaceView.setRenderer { frame ->
//     queue.enqueue(frame.timestamp)
//     // 交换 EGL Buffer → encoder inputSurface
//     eglSwapBuffers()
// }

4、异步编码五大陷阱

4.1、陷阱一：忘记 releaseOutputBuffer

// ❌ 最常见的泄漏
overridefun onOutputBufferAvailable(codec: MediaCodec, index: Int, info: BufferInfo) {
    val buffer = codec.getOutputBuffer(index)
    mediaMuxer?.writeSampleData(videoTrackIndex, buffer!!, info)
    // 忘了 releaseOutputBuffer！Buffer 池耗尽 → 编码器卡死
}

// ✅ 每次必须 release
overridefun onOutputBufferAvailable(codec: MediaCodec, index: Int, info: BufferInfo) {
    val buffer = codec.getOutputBuffer(index)
    try {
        mediaMuxer?.writeSampleData(videoTrackIndex, buffer!!, info)
    } finally {
        codec.releaseOutputBuffer(index, false)  // ← 必须调用
    }
}

4.2、陷阱二：在错误线程操作编码器

// ❌ 在主线程调用 codec.stop() — ANR
// ✅ 所有操作在 codecHandler 所在线程
codecHandler.post {
    codec.stop()
    codec.release()
}

4.3、陷阱三：Muxer 时序错误

// ❌ Muxer.start() 前 writeSampleData → crash
// ✅ 在 onOutputFormatChanged 中 start Muxer，之后才写入
override fun onOutputFormatChanged(codec: MediaCodec, format: MediaFormat) {
    videoTrackIndex = muxer.addTrack(format)
    muxer.start()
    muxerStarted = true  // 之后 onOutputBufferAvailable 才写入
}

4.4、陷阱四：EOS 处理不当

// ❌ 直接 stop → 最后几帧丢失
mediaCodec.stop()

// ✅ 先 signalEndOfInputStream → 等待 EOS 回调 → 再 stop
mediaCodec.signalEndOfInputStream()
// 等待 onOutputBufferAvailable 收到 BUFFER_FLAG_END_OF_STREAM
// 然后 mediaCodec.stop()

4.5、陷阱五：重建编码器时未清理

// 编码器异常 → 自动重建
override fun onError(codec: MediaCodec, e: MediaCodec.CodecException) {
    // 1. 释放旧编码器
    codec.release()
    // 2. 旧 Muxer 也需重建（因为 track index 变了）
    mediaMuxer?.stop()
    mediaMuxer?.release()
    // 3. 重新创建
    scope?.launch { configureAndStart() }
}

5、性能数据

学习和提升音视频开发技术，欢迎你加入我们的知识星球