基于go-rtmp的RTMP服务器音频0.5秒批量推送卡顿问题排查

问题背景 #

使用go-rtmp搭建的RTMP服务器，原OnAudio函数每0.02秒接收并推送一次音频数据，运行正常。修改为累计0.5秒音频数据后再推送，数据无丢失但播放卡顿。

原OnAudio函数代码 #

func (h *Handler) OnAudio(timestamp uint32, payload io.Reader) error {
    var audio flvtag.AudioData
    if err := flvtag.DecodeAudioData(payload, &audio); err != nil {
        return err
    }

    flvBody := new(bytes.Buffer)
    if _, err := io.Copy(flvBody, audio.Data); err != nil {
        return err
    }
    audio.Data = flvBody
    _ = h.pub.Publish(&flvtag.FlvTag{
        TagType:   flvtag.TagTypeAudio,
        Timestamp: timestamp,
        Data:      &audio,
    })

    return nil
}

修改后的OnAudio函数代码 #

func (h *Handler) OnAudio(timestamp uint32, payload io.Reader) error {

    var audio flvtag.AudioData
    if err := flvtag.DecodeAudioData(payload, &audio); err != nil {
        return err
    }
    // if flag == true, startTimeStamp Control
    if h.Flag == true {
        h.startTimeStamp = timestamp
        h.Buf.Reset()
    }
    h.Flag = false

    buf := new(bytes.Buffer)
    if _, err := io.Copy(buf, audio.Data); err != nil {
        return err
    }

    if audio.AACPacketType == 0 {
        audio.Data = buf
        tag := &flvtag.FlvTag{
            TagType:   flvtag.TagTypeAudio,
            Timestamp: h.startTimeStamp,
            Data:      &audio,
        }
        // 注：原代码此处有语法错误，已保留原样
        timestamp, (timestamp - h.startTimeStamp), buf.Len())
        _ = h.pub.Publish(tag)
        h.Flag = true
        return nil
    }

    h.Buf.Write(buf.Bytes())

    // collect audio data into h.Buf and then publish every 0.5 seconds
    if timestamp-h.startTimeStamp >= 500 {
        h.lastTimeStamp = timestamp
        audio.Data = h.Buf
        tag := &flvtag.FlvTag{
            TagType:   flvtag.TagTypeAudio,
            Timestamp: h.startTimeStamp,
            Data:      &audio,
        }

        _ = h.pub.Publish(tag)
        h.Flag = true
    }

    return nil
}

问题分析 #

卡顿的核心原因是破坏了FLV音频Tag的规范结构，具体问题点如下：

• FLV格式违规：FLV的Audio Tag要求每个Tag对应单个可解析的音频单元（比如单个AAC帧）。你直接将多个音频帧的原始字节合并到一个Tag的Data字段中，播放器无法识别这种非标准格式，解析时会出现时序混乱或解码失败，导致卡顿。
• 时间戳映射错误：合并后的Tag使用初始时间戳，但每个音频帧都有自己对应的播放时间戳，单个Tag的时间戳无法匹配多个帧的时序，播放器无法正确同步播放节奏。
• 剩余数据隐患：推流结束时，缓存中未达500ms的音频数据会被丢弃（当前数据无丢失是因为推流未中断，但长期运行会有数据丢失风险）。

修正方案 #

不要合并音频帧的原始数据，而是缓存完整的FLV Audio Tag，累计到0.5秒时长后批量推送，保留每个Tag的原始时间戳和结构。

修正后的代码示例 #

type Handler struct {
    pub            *rtmp.Publisher
    audioTagCache  []*flvtag.FlvTag
    startTimeStamp uint32
}

func (h *Handler) OnAudio(timestamp uint32, payload io.Reader) error {
    var audio flvtag.AudioData
    if err := flvtag.DecodeAudioData(payload, &audio); err != nil {
        return err
    }

    // 复制音频数据到独立Buffer，避免原Reader被复用消耗
    buf := new(bytes.Buffer)
    if _, err := io.Copy(buf, audio.Data); err != nil {
        return err
    }
    audio.Data = buf

    // 创建符合规范的FLV音频Tag
    tag := &flvtag.FlvTag{
        TagType:   flvtag.TagTypeAudio,
        Timestamp: timestamp,
        Data:      &audio,
    }

    // AAC序列头需要立即推送，不参与缓存
    if audio.AACPacketType == 0 {
        return h.pub.Publish(tag)
    }

    // 初始化缓存的起始时间戳
    if len(h.audioTagCache) == 0 {
        h.startTimeStamp = timestamp
    }

    // 缓存当前完整的FLV Tag
    h.audioTagCache = append(h.audioTagCache, tag)

    // 检查累计时长是否达到500ms，达到则批量推送
    if timestamp-h.startTimeStamp >= 500 {
        for _, cachedTag := range h.audioTagCache {
            if err := h.pub.Publish(cachedTag); err != nil {
                return err
            }
        }
        // 清空缓存，准备下一轮累计
        h.audioTagCache = nil
    }

    return nil
}

// 补充：在推流断开时推送剩余缓存的音频Tag
func (h *Handler) OnClose() {
    if len(h.audioTagCache) > 0 {
        for _, cachedTag := range h.audioTagCache {
            _ = h.pub.Publish(cachedTag)
        }
        h.audioTagCache = nil
    }
}

修正逻辑说明 #

1. 每个音频帧都保留为独立的FLV Audio Tag，符合FLV规范，播放器能正确解析每个帧的格式和时间戳。
2. 批量推送只是减少了网络请求次数，不会破坏音频的播放时序，解决卡顿问题。
3. 补充OnClose方法处理剩余缓存，避免推流结束时丢失最后一段音频数据。

内容的提问来源于stack exchange，提问作者Kundera