goroutine通道中的同步问题

在Golang实战开发的过程中，我们经常会遇到一些这样那样的问题，然后要卡好半天，等问题解决了才发现原来一些细节知识点还是没有掌握好。今天golang学习网就整理分享《goroutine通道中的同步问题》，聊聊，希望可以帮助到正在努力赚钱的你。

我正在尝试编写一个代理服务器，在其中将视频文件转换为实时流。一个视频点播文件由多个不同比特率的子清单文件组成。每个子清单由多个 ts 段组成，每个 ts 段为 4 秒。为简单起见，我创建了 2 个子清单的虚拟映射，每个子清单包含 4 个 ts 段。我的任务是在 for 循环中无限创建直播流。所以我在 go 例程中 4 秒后发送每个 ts 段并写入通道，并且我还有一个 go 例程，我在其中读取此 go 例程并写入输出字符串。一旦所有片段结束，我将更新到 loopend 频道并重新开始。有 2 个针对不同比特率的 api，每 4 秒后给我一个新的 ts。

package main

import (
    "fmt"
    "github.com/gin-contrib/gzip"
    "github.com/gin-gonic/gin"
    "sync"
    "time"
)

func main() {
    vodToLiveMap := convertVodToLive()
    engine := gin.Default()
    engine.Use(gzip.Gzip(gzip.DefaultCompression))
    engine.GET("/134200", func(gctx *gin.Context) {
        prepareResponse(gctx, "134200", vodToLiveMap)
    })

    engine.GET("/290400", func(gctx *gin.Context) {
        prepareResponse(gctx, "290400", vodToLiveMap)
    })
    engine.Run() // listen and serve on 0.0.0.0:8080
}

// this represents one ulr for http live stream
type hlsEntry struct {
    timeCode int    // timecode in microseconds
    body     string // string including transport segment URL, timecode
}

type VodToLive struct {
    SequenceNumber int
    Out            chan string
    CurrentStream  string
}

func convertVodToLive() sync.Map {
    var hlsEntriesBitrate1 = []hlsEntry{
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer1_001.ts\n",
        },
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer1_002.ts\n",
        },
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer1_003.ts\n",
        },
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer1_004.ts\n",
        },
    }

    var hlsEntriesBitrate2 = []hlsEntry{
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer2_001.ts\n",
        },
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer2_002.ts\n",
        },
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer2_003.ts\n",
        },
        {
            timeCode: 4000000,
            body:     "#EXTINF:4.00000,\n/master_Layer2_004.ts\n",
        },
    }
    fmt.Println(hlsEntriesBitrate2)
    var hlsEntriesMap = make(map[string][]hlsEntry)
    hlsEntriesMap["134200"] = hlsEntriesBitrate1
    hlsEntriesMap["290400"] = hlsEntriesBitrate2
    outputString := ""
    vodToLiveMap := sync.Map{}
    i := 0
    //Iterate for all child manifest files of different bit rates
    for childURLSlug := range hlsEntriesMap {
        vodToLive := &VodToLive{}
        vodToLive.SequenceNumber = 1
        vodToLive.Out = make(chan string)
        vodToLiveMap.Store(childURLSlug, vodToLive)
        loopEnd := make(chan bool)

        //Writing to vodToLive for a childURLSlug
        go func(childURLSlug string) {
            for {
                for _, entry := range hlsEntriesMap[childURLSlug] {
                    select {
                    case <-time.After(time.Duration(entry.timeCode) * time.Microsecond):
                        vodToLive.SequenceNumber++
                        vodToLive.Out <- entry.body
                    }
                }
                loopEnd <- true
                fmt.Println(childURLSlug)
            }
        }(childURLSlug)

        //Reading from vodToLive's Out channel for a childURLSlug to update CurrentStream
        go func(childURLSlug string) {
            for {
                for {
                    select {
                    //If video reaches end of stream then restart stream from Sequence #1
                    case <-loopEnd:
                        vodToLive.SequenceNumber = 1
                        //time.Sleep(10 * time.Second)
                        vodToLive.CurrentStream = ""
                        break

                    default:
                        outputString = <-vodToLive.Out
                        fmt.Println(outputString)
                        vodToLive.CurrentStream = vodToLive.CurrentStream + outputString

                    }
                }
            }
        }(childURLSlug)
        i++
    }
    return vodToLiveMap
}

func prepareResponse(gctx *gin.Context, childURLSlug string, vodToLiveMap sync.Map) {

    vodToLive, _ := vodToLiveMap.Load(childURLSlug)

    gctx.Header("Content-Type", "application/x-mpegURL")
    response := `#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:4
#EXT-X-MEDIA-SEQUENCE:` + "1" + "\n" + vodToLive.(*VodToLive).CurrentStream

    size := len(response)
    if size > 0 && response[size-1] == '\n' {
        response = response[:size-1]
    }
    gctx.String(200, response)
}
//curl localhost:8080/134200
//curl localhost:8080/290400

但这里的问题是我没有在curl请求中获取所有ts段，有时甚至ts在循环结束后也没有更新。我在每个调用中显示了所有 4 个 ts 段，但这些应该在每个 endloop 后重置。

解决方案

只需将默认值替换为通道上的选择 (vodtolive.out) 并在 case <-loopend 中添加 4 秒等待即可解决此问题

go func(childURLSlug string) {
    for {
        for {
            select {
            //If video reaches end of stream then restart stream from Sequence #1
            case <-loopEnd:
                vodToLive.SequenceNumber = 1
                // so this will ensure that next stream starts only when first is complete.
                time.Sleep(4 * time.Second) //Usually a ts is 4 seconds length.
                vodToLive.CurrentStream = ""
                break

            case res := <-vodToLive.Out:
                outputString := res
                fmt.Println(outputString)
                vodToLive.CurrentStream = vodToLive.CurrentStream + outputString

            }
        }
    }
}(childURLSlug)

理论要掌握，实操不能落！以上关于《goroutine通道中的同步问题》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！