02-05-08 直播推流与拉流

1. 概述

直播技术是音视频领域的重要应用，包括推流（采集编码推送）和拉流（拉取解码播放）两部分。主流协议有RTMP、HLS、HTTP-FLV、WebRTC等。

2. RTMP推流

2.1 推流架构

Camera/Mic → MediaCodec编码 → RTMP封装 → 推送到服务器

2.2 推流实现

class RtmpPusher(private val rtmpUrl: String) {
    private lateinit var videoEncoder: MediaCodec
    private lateinit var audioEncoder: MediaCodec
    private var rtmpClient: RtmpClient? = null

    fun startPush() {
        // 1. 初始化编码器
        initVideoEncoder()
        initAudioEncoder()

        // 2. 连接RTMP服务器
        rtmpClient = RtmpClient(rtmpUrl)
        rtmpClient?.connect()

        // 3. 开始推流
        startVideoEncoding()
        startAudioEncoding()
    }

    private fun initVideoEncoder() {
        videoEncoder = MediaCodec.createEncoderByType(MediaFormat.MIMETYPE_VIDEO_AVC)
        val format = MediaFormat().apply {
            setString(MediaFormat.KEY_MIME, MediaFormat.MIMETYPE_VIDEO_AVC)
            setInteger(MediaFormat.KEY_WIDTH, 1280)
            setInteger(MediaFormat.KEY_HEIGHT, 720)
            setInteger(MediaFormat.KEY_BIT_RATE, 2_000_000)
            setInteger(MediaFormat.KEY_FRAME_RATE, 30)
            setInteger(MediaFormat.KEY_I_FRAME_INTERVAL, 2)
        }
        videoEncoder.configure(format, null, null, MediaCodec.CONFIGURE_FLAG_ENCODE)
        videoEncoder.start()
    }

    private fun startVideoEncoding() {
        Thread {
            val bufferInfo = MediaCodec.BufferInfo()

            while (true) {
                val outputBufferId = videoEncoder.dequeueOutputBuffer(bufferInfo, 10_000)

                if (outputBufferId >= 0) {
                    val outputBuffer = videoEncoder.getOutputBuffer(outputBufferId)
                    outputBuffer?.let {
                        val data = ByteArray(bufferInfo.size)
                        it.get(data)

                        // 推送到RTMP服务器
                        rtmpClient?.sendVideoData(data, bufferInfo.presentationTimeUs,
                            (bufferInfo.flags and MediaCodec.BUFFER_FLAG_KEY_FRAME) != 0)

                        videoEncoder.releaseOutputBuffer(outputBufferId, false)
                    }
                }
            }
        }.start()
    }

    fun stopPush() {
        videoEncoder.stop()
        videoEncoder.release()
        audioEncoder.stop()
        audioEncoder.release()
        rtmpClient?.disconnect()
    }
}

// RTMP客户端封装（伪代码）
class RtmpClient(private val url: String) {
    private var socket: Socket? = null

    fun connect() {
        // 解析RTMP URL
        // 建立TCP连接
        // RTMP握手
        // 发送connect命令
    }

    fun sendVideoData(data: ByteArray, timestamp: Long, isKeyFrame: Boolean) {
        // 封装RTMP视频包
        // 发送到服务器
    }

    fun sendAudioData(data: ByteArray, timestamp: Long) {
        // 封装RTMP音频包
        // 发送到服务器
    }

    fun disconnect() {
        socket?.close()
    }
}

3. HLS拉流

3.1 HLS播放器

class HlsPlayer(private val context: Context) {
    private lateinit var exoPlayer: ExoPlayer

    fun play(hlsUrl: String) {
        exoPlayer = ExoPlayer.Builder(context).build()

        val mediaItem = MediaItem.Builder()
            .setUri(hlsUrl)
            .setMimeType(MimeTypes.APPLICATION_M3U8)
            .build()

        exoPlayer.setMediaItem(mediaItem)
        exoPlayer.prepare()
        exoPlayer.play()
    }

    fun stop() {
        exoPlayer.release()
    }
}

3.2 自定义HLS解析

class HlsParser {
    data class HlsPlaylist(
        val segments: List<Segment>,
        val targetDuration: Int,
        val mediaSequence: Int
    )

    data class Segment(
        val duration: Float,
        val url: String
    )

    fun parse(m3u8Content: String): HlsPlaylist {
        val lines = m3u8Content.lines()
        val segments = mutableListOf<Segment>()
        var targetDuration = 0
        var mediaSequence = 0
        var currentDuration = 0f

        for (line in lines) {
            when {
                line.startsWith("#EXT-X-TARGETDURATION:") -> {
                    targetDuration = line.substringAfter(":").toInt()
                }
                line.startsWith("#EXT-X-MEDIA-SEQUENCE:") -> {
                    mediaSequence = line.substringAfter(":").toInt()
                }
                line.startsWith("#EXTINF:") -> {
                    currentDuration = line.substringAfter(":").substringBefore(",").toFloat()
                }
                !line.startsWith("#") && line.isNotBlank() -> {
                    segments.add(Segment(currentDuration, line))
                }
            }
        }

        return HlsPlaylist(segments, targetDuration, mediaSequence)
    }
}

4. HTTP-FLV拉流

class FlvPlayer(private val flvUrl: String) {
    private lateinit var decoder: MediaCodec
    private var isPlaying = false

    fun play() {
        isPlaying = true

        Thread {
            // 1. 连接HTTP服务器
            val connection = URL(flvUrl).openConnection() as HttpURLConnection
            val inputStream = connection.inputStream

            // 2. 解析FLV头部
            val flvHeader = ByteArray(9)
            inputStream.read(flvHeader)

            // 3. 循环读取FLV Tag
            while (isPlaying) {
                val tag = readFlvTag(inputStream)
                when (tag.type) {
                    8 -> processAudioTag(tag)  // 音频
                    9 -> processVideoTag(tag)  // 视频
                }
            }
        }.start()
    }

    private fun readFlvTag(inputStream: InputStream): FlvTag {
        // 读取FLV Tag
        val tagType = inputStream.read()
        val dataSize = readInt24(inputStream)
        val timestamp = readInt24(inputStream)
        val data = ByteArray(dataSize)
        inputStream.read(data)

        return FlvTag(tagType, data, timestamp.toLong())
    }

    private fun processVideoTag(tag: FlvTag) {
        // 解码并渲染视频帧
    }

    data class FlvTag(val type: Int, val data: ByteArray, val timestamp: Long)

    private fun readInt24(input: InputStream): Int {
        return (input.read() shl 16) or (input.read() shl 8) or input.read()
    }
}

5. 低延迟优化

5.1 关键帧请求

class LowLatencyOptimizer(private val encoder: MediaCodec) {
    // 网络抖动时请求关键帧
    fun requestKeyFrame() {
        val bundle = Bundle().apply {
            putInt(MediaCodec.PARAMETER_KEY_REQUEST_SYNC_FRAME, 0)
        }
        encoder.setParameters(bundle)
    }

    // 动态码率调整
    fun adjustBitrate(newBitrate: Int) {
        val bundle = Bundle().apply {
            putInt(MediaCodec.PARAMETER_KEY_VIDEO_BITRATE, newBitrate)
        }
        encoder.setParameters(bundle)
    }
}

5.2 缓冲区管理

class BufferManager {
    private val bufferQueue = LinkedBlockingQueue<ByteArray>(10)

    // 限制缓冲区大小
    fun addBuffer(data: ByteArray) {
        if (!bufferQueue.offer(data)) {
            // 缓冲区满，丢弃最旧的帧
            bufferQueue.poll()
            bufferQueue.offer(data)
        }
    }

    fun getBuffer(): ByteArray? {
        return bufferQueue.poll()
    }

    fun getBufferSize(): Int = bufferQueue.size
}

6. 案例研究

案例1：斗鱼直播技术方案

推流端：

• OBS推流工具
• RTMP协议
• H.264编码，码率自适应

服务器：

• RTMP源站接收
• 转码成多码率
• 分发到CDN

播放端：

• HLS（延迟10-30秒）
• HTTP-FLV（延迟2-5秒）
• WebRTC（延迟<1秒）

案例2：抖音直播优化

技术方案：

• 硬件编码（MediaCodec）
• 自研传输协议（基于QUIC）
• AI美颜实时处理
• 动态码率调整

延迟优化：

• GOP缩短至1秒
• 服务端缓冲500ms
• 端到端延迟<2秒

---

总结

直播技术涵盖采集、编码、推流、拉流、解码、渲染全链路：

1. RTMP推流：主流推流协议
2. HLS拉流：兼容性好，延迟高
3. HTTP-FLV：延迟低，兼容性好
4. 低延迟优化：关键帧请求、缓冲区管理

关键要点：

• 硬件编码降低功耗
• 动态码率适应网络
• 协议选择平衡延迟与兼容性

下一篇预告：《WebRTC实时通信》