Go-如何处理Worker填写自己的jobQueue

如何解决Go-如何处理Worker填写自己的jobQueue

作为Go中的新手，我目前正在实现一个网络爬虫，以从在线视频游戏中获取比赛信息。 但是，我正在努力充分理解如何正确使用goroutines以及如何有效处理通道。

从本质上讲，我希望“工人”（“ CrawlPlayers”）填写自己的jobQueue（“玩家”）。至关重要的是，对于工人从渠道获得的每个工作，渠道本身都会再获得1-9个新工作。

游戏的一些环境：

• 每场比赛由10位玩家（5v5）组成。
• 每个球员都有他们参加过的比赛的列表（“ MatchList”）

每个“抓取工具”对象都有一个“开始”功能，该功能在获取100个匹配项后会中止：

// Void is a shortcut type for struct{} that is specifically used for maps
type Void struct{}

type Crawler struct {
    ...
    startPlayer string
    store       *Store
    playerChan  chan string
    quit        chan Void
}

func NewCrawler(startPlayer string) *Crawler {
    return &Crawler{
        ...
        startPlayer: startPlayer,store:       NewStore(),// object that entails a threadsafe map for storing processed matches
        playerChan:  make(chan string,100),quit:        make(chan Void),}
}

func (c *Crawler) Start() {
    defer close(c.playerChan)
    defer close(c.quit)
    sp,err := c.GetPlayerByName(c.startPlayer)
    if err != nil {
        fmt.Println("Erronous Start Player given!")
        return
    }

    for i := 1; i <= runtime.NumCPU(); i++ {
        go c.CrawlPlayer(c.playerChan,c.quit)
    }

    go func() {
        for {
            if c.store.NumMatches() >= 100 {
                c.quit <- Void{}
                return
            }
        }
    }()
    c.playerChan <- sp.AccountId
    <-c.quit
    fmt.Println("finished")
}

CrawlPlayer函数从其专用频道接收某个玩家，查找其MatchList并从中获取所有可用的比赛

func (c *Crawler) CrawlPlayer(players chan string,quit <-chan Void) {
   select {
    case <-quit:
        close(players)
        return
    case <-time.After(1 * time.Minute):
        fmt.Println("Timeout after 1 Minute,crawling next player")
        return
    case player := <-players:
        // 1. Get the Matchlist
        ...
        // 2. Process each match
        for _,m := range ml.Matches {
            // 2.1 store the match object
            ...
            // 2.2 Get the next players to crawl matches from
            for _,part := range match.participants {
                aID := part.playerId
                if c.store.PlayerExists(aID) {
                    fmt.Println("Player already crawled",aID)
                    continue
                }
                players <- aID
            }
        }
        fmt.Printf("Finished working on player %s",player)
    }

}

该问题出现在第2.2节中。 在11场比赛之后，玩家通道只会被阻塞，应用程序将不会前进。

但是，当我创建一个新的缓冲通道并使用该新通道启动新的goroutine时，该应用程序似乎至少要在11个匹配项后恢复。

// 2.2 Get the next players to crawl matches from
            newPlayers := make(chan string,10)
            defer close(newPlayers)
            go c.CrawlPlayer(newPlayers,quit)
            for _,pID := range match.participants {
                aID := pID.Player.AccountId
                if c.store.PlayerExists(aID) {
                    fmt.Println("Player already crawled",aID)
                    continue
                }
                newPlayers <- aID
           }

但是：

1. 这种方法对我来说意义不大，因为我想利用原始播放器通道和我在Start（）函数中开始的goroutines

2. 由于goroutine呈指数增长，因此是否存在内存泄漏问题？每场比赛，我都会吸引至少9名球员和他们自己的比赛清单。