如何解决Go中HLS流下载合并MP3的测试文件不一致问题？

原下载函数代码 #

func DownloadM3u8(filepath string, dlbar *bar.ProgressBar, segments []string) error {

    file, _ := os.OpenFile(filepath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)

    // 启动goroutine
    var wg sync.WaitGroup

    for _, segment := range segments {
        wg.Add(1)
        downloadSeg(&wg, segment, file, dlbar)
    }

    return nil
}


func downloadSeg(wg *sync.WaitGroup, segmentURI string, file *os.File, dlbar *bar.ProgressBar) {
    defer wg.Done()
    resp, err := http.Get(segmentURI)

    if err != nil {
        return
    }

    defer resp.Body.Close()

    // 写入文件
    if dlbar == nil {
        _, err = io.Copy(io.MultiWriter(file), resp.Body)
    } else {
        _, err = io.Copy(io.MultiWriter(file, dlbar), resp.Body)
    }

    if err != nil {
        return
    }

}

传入的segments为URI列表，示例格式：

[
https://url-adc.com/s/933030/30032, 
https://url-adc.com/s/933030/303220,
https://url-adc.com/s/933030/34230,
https://url-adc.com/s/933030/35290,
]

测试思路 #

使用已下载完成的Test[medium].mp3文件，将其分割为[]byte片段并绑定到模拟URL，通过httptest包返回对应字节数据，验证合并后的文件与原文件是否一致。测试函数代码：

func TestDownloadM3u8(t *testing.T) {

    path := "../../testdata/"
    track := readTestFile("Test[medium].mp3") // 返回[]byte

    fileName := "some filename that exists" // 预期生成的临时mp3文件名
    path = filepath.Join(path, fileName)
       // 从实际m3u8文件提取分段信息
    segments := extractSegments(fileResp, track)

    // 启动测试服务器模拟分段下载
    testServer := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
        // 根据URL参数返回对应分段字节
        n, _ := strconv.Atoi(req.URL.String()[1:])
        if bs, ok := segments[n]; ok {
            res.WriteHeader(http.StatusOK)
            res.Write(bs)
        }
    }))

    defer testServer.Close()

    segmentURIs := make([]string, 0)
    for k := range segments {
        segmentURIs = append(segmentURIs, testServer.URL+"/"+strconv.Itoa(k))
    }
    soundcloud.DownloadM3u8(path, nil, segmentURIs)

    // 读取下载后的文件
    file, err := ioutil.ReadFile(path)
    if err != nil {
        t.Errorf("读取文件时出错: %s", err)
    }
    // 对比文件内容
    if !bytes.Equal(file, track) {
        t.Errorf("预期两个文件内容一致，但实际不一致")
    }
    // os.Remove(path) // 可选：删除测试文件
}

分段提取与分割逻辑 #

extractSegments函数读取m3u8播放列表获取分段数量，再分割测试文件：

func extractSegments(fileP []byte, testfile []byte) map[int][]byte {
    segments := make(map[int][]byte, 0)
    reader := bytes.NewReader(fileP)

    pl, listType, err := m3u8.DecodeFrom(reader, true)

    if err != nil {
        return nil
    }

    switch listType {
    case m3u8.MEDIA:
        mediapl := pl.(*m3u8.MediaPlaylist)
        for i, segment := range mediapl.Segments {
            if segment == nil {
                continue
            }
                     // 此处需要将testfile分割为对应片段
            segments[i] = ...
        }
    }

    return segments
}

最初的分割逻辑：

var segmentSize int
    numSegments := len(segments)

    fileSize := len(testfile)
    segmentSize = fileSize / numSegments

    for i := 0; i < numSegments-1; i++ {
        start := i * segmentSize
        end := (i + 1) * segmentSize
        segments[i] = testfile[start:end]
    }

    // 处理最后一段
    start := (numSegments - 1) * segmentSize
    end := fileSize
    if fileSize%numSegments == 0 {
        end = (numSegments-1)*segmentSize + segmentSize
    }
    segments[int(numSegments-1)] = testfile[start:end]

测试后发现合并文件与原文件不一致，需调整代码。

问题原因与修复方案 #

核心问题1：并发写入导致顺序混乱 #

原下载函数用goroutine并行写入文件，但os.File的Write操作并非并发安全，且goroutine执行顺序不确定，会导致分段内容乱序写入，最终文件字节顺序与原文件不符。

核心问题2：分割逻辑的小瑕疵 #

原分割逻辑对余数的处理冗余，虽不是主要问题，但可优化避免潜在的字节遗漏。

修复步骤 #

步骤1：保证分段写入的顺序性

提供两种方案：

• 方案A（测试场景快速修复）：取消并行，按顺序下载写入
• 方案B（生产级并发安全）：并行下载到内存缓冲区，再按顺序写入文件

方案A代码修改：

func DownloadM3u8(filepath string, dlbar *bar.ProgressBar, segments []string) error {
    // 不要忽略文件打开错误
    file, err := os.OpenFile(filepath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
    if err != nil {
        return err
    }
    defer file.Close() // 必须关闭文件，确保内容刷入磁盘

    // 按顺序下载每个分段
    for _, segment := range segments {
        downloadSeg(nil, segment, file, dlbar)
    }

    return nil
}

// 调整downloadSeg，移除WaitGroup相关逻辑
func downloadSeg(_ *sync.WaitGroup, segmentURI string, file *os.File, dlbar *bar.ProgressBar) {
    resp, err := http.Get(segmentURI)
    if err != nil {
        return
    }
    defer resp.Body.Close()

    var writer io.Writer = file
    if dlbar != nil {
        writer = io.MultiWriter(file, dlbar)
    }
    _, err = io.Copy(writer, resp.Body)
    if err != nil {
        return
    }
}

方案B代码修改：

type segmentData struct {
    index int
    data  []byte
    err   error
}

func DownloadM3u8(filepath string, dlbar *bar.ProgressBar, segments []string) error {
    file, err := os.OpenFile(filepath, os.O_CREATE|os.O_WRONLY, 0644)
    if err != nil {
        return err
    }
    defer file.Close()

    var wg sync.WaitGroup
    dataChan := make(chan segmentData, len(segments))

    // 并行下载所有分段到内存
    for i, segment := range segments {
        wg.Add(1)
        go func(idx int, uri string) {
            defer wg.Done()
            resp, err := http.Get(uri)
            if err != nil {
                dataChan <- segmentData{index: idx, err: err}
                return
            }
            defer resp.Body.Close()

            data, err := io.ReadAll(resp.Body)
            if err != nil {
                dataChan <- segmentData{index: idx, err: err}
                return
            }

            dataChan <- segmentData{index: idx, data: data}
        }(i, segment)
    }

    // 等待所有下载完成后关闭通道
    go func() {
        wg.Wait()
        close(dataChan)
    }()

    // 收集并按原始索引排序分段
    segmentList := make([]segmentData, 0, len(segments))
    for sd := range dataChan {
        segmentList = append(segmentList, sd)
    }

    sort.Slice(segmentList, func(i, j int) bool {
        return segmentList[i].index < segmentList[j].index
    })

    // 检查下载错误
    for _, sd := range segmentList {
        if sd.err != nil {
            return sd.err
        }
    }

    // 按顺序写入文件
    totalWritten := 0
    for _, sd := range segmentList {
        n, err := file.Write(sd.data)
        if err != nil {
            return err
        }
        totalWritten += n
        if dlbar != nil {
            dlbar.Add(n)
        }
    }

    return nil
}

步骤2：优化分割逻辑

将余数均匀分配给前几个分段，确保所有字节都被正确分割：

func extractSegments(fileP []byte, testfile []byte) map[int][]byte {
    segments := make(map[int][]byte)
    reader := bytes.NewReader(fileP)

    pl, listType, err := m3u8.DecodeFrom(reader, true)
    if err != nil {
        return nil
    }

    switch listType {
    case m3u8.MEDIA:
        mediapl := pl.(*m3u8.MediaPlaylist)
        numSegments := len(mediapl.Segments)
        if numSegments == 0 {
            return segments
        }
        fileSize := len(testfile)
        segmentSize := fileSize / numSegments
        remaining := fileSize % numSegments

        current := 0
        for i, segment := range mediapl.Segments {
            if segment == nil {
                continue
            }
            end := current + segmentSize
            // 前remaining个分段多分配1字节，处理余数
            if i < remaining {
                end += 1
            }
            // 防止越界
            if end > fileSize {
                end = fileSize
            }
            segments[i] = testfile[current:end]
            current = end
        }
    }

    return segments
}

步骤3：测试函数优化

使用临时文件避免覆盖已有文件，同时完善错误处理：

func TestDownloadM3u8(t *testing.T) {
    track := readTestFile("Test[medium].mp3")
    if track == nil {
        t.Fatal("无法读取测试文件")
    }

    // 创建临时文件
    tmpFile, err := os.CreateTemp("", "test-m3u8-*.mp3")
    if err != nil {
        t.Fatalf("创建临时文件失败: %s", err)
    }
    tmpPath := tmpFile.Name()
    defer func() {
        tmpFile.Close()
        os.Remove(tmpPath) // 测试结束后自动删除
    }()

    segments := extractSegments(fileResp, track)
    if segments == nil {
        t.Fatal("提取分段失败")
    }

    testServer := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
        n, err := strconv.Atoi(req.URL.String()[1:])
        if err != nil {
            res.WriteHeader(http.StatusBadRequest)
            return
        }
        if bs, ok := segments[n]; ok {
            res.WriteHeader(http.StatusOK)
            res.Write(bs)
        } else {
            res.WriteHeader(http.StatusNotFound)
        }
    }))
    defer testServer.Close()

    // 按索引顺序生成URI，确保请求顺序正确
    segmentURIs := make([]string, 0, len(segments))
    for i := 0; i < len(segments); i++ {
        if _, ok := segments[i]; ok {
            segmentURIs = append(segmentURIs, testServer.URL+"/"+strconv.Itoa(i))
        }
    }

    err = soundcloud.DownloadM3u8(tmpPath, nil, segmentURIs)
    if err != nil {
        t.Fatalf("下载合并失败: %s", err)
    }

    downloaded, err := ioutil.ReadFile(tmpPath)
    if err != nil {
        t.Fatalf("读取下载文件失败: %s", err)
    }

    if !bytes.Equal(downloaded, track) {
        t.Errorf("文件内容不一致，原文件长度%d，下载文件长度%d", len(track), len(downloaded))
    }
}

总结 #

测试失败的核心原因是并发写入导致分段顺序混乱，修复后确保分段按原始顺序写入文件，配合优化后的分割逻辑，即可让合并后的文件与原文件完全一致。

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