這是我第一次使用 Go 的并發特性，我正在深入研究。
我想對 API 進行并發呼叫。該請求基于我想收到的帖子的標簽（可以有 1 <= N 個標簽）。回應正文如下所示：
{
"posts": [
{
"id": 1,
"author": "Name",
"authorId": 1,
"likes": num_likes,
"popularity": popularity_decimal,
"reads": num_reads,
"tags": [ "tag1", "tag2" ]
},
...
]
}
我的計劃是將一堆通道菊花鏈在一起，并產生許多從這些通道讀取和/或寫入的 goroutine：
- for each tag, add it to a tagsChannel inside a goroutine
- use that tagsChannel inside another goroutine to make concurrent GET requests to the endpoint
- for each response of that request, pass the underlying slice of posts into another goroutine
- for each individual post inside the slice of posts, add the post to a postChannel
- inside another goroutine, iterate over postChannel and insert each post into a data structure
這是我到目前為止所擁有的：
func (srv *server) Get() {
// Using red-black tree prevents any duplicates, fast insertion
// and retrieval times, and is sorted already on ID.
rbt := tree.NewWithIntComparator()
// concurrent approach
tagChan := make(chan string) // tags -> tagChan
postChan := make(chan models.Post) // tagChan -> GET -> post -> postChan
errChan := make(chan error) // for synchronizing errors across goroutines
wg := &sync.WaitGroup{} // for synchronizing goroutines
wg.Add(4)
// create a go func to synchronize our wait groups
// once all goroutines are finished, we can close our errChan
go func() {
wg.Wait()
close(errChan)
}()
go insertTags(tags, tagChan, wg)
go fetch(postChan, tagChan, errChan, wg)
go addPostToTree(rbt, postChan, wg)
for err := range errChan {
if err != nil {
srv.HandleError(err, http.StatusInternalServerError).ServeHTTP(w, r)
}
}
}
// insertTags inserts user's passed-in tags to tagChan
// so that tagChan may pass those along in fetch.
func insertTags(tags []string, tagChan chan<- string, group *sync.WaitGroup) {
defer group.Done()
for _, tag := range tags {
tagChan <- tag
}
close(tagChan)
}
// fetch completes a GET request to the endpoint
func fetch(posts chan<- models.Post, tags <-chan string, errs chan<- error, group *sync.WaitGroup) {
defer group.Done()
for tag := range tags {
ep, err := formURL(tag)
if err != nil {
errs <- err
}
group.Add(1) // QUESTION should I use a separate wait group here?
go func() {
resp, err := http.Get(ep.String())
if err != nil {
errs <- err
}
container := models.PostContainer{}
err = json.NewDecoder(resp.Body).Decode(&container)
defer resp.Body.Close()
group.Add(1) // QUESTION should I add a separate wait group here and pass it to insertPosts?
go insertPosts(posts, container.Posts, group)
defer group.Done()
}()
// group.Done() -- removed this call due to Burak, but now my program hands
}
}
// insertPosts inserts each individual post into our posts channel so that they may be
// concurrently added to our RBT.
func insertPosts(posts chan<- models.Post, container []models.Post, group *sync.WaitGroup) {
defer group.Done()
for _, post := range container {
posts <- post
}
}
// addPostToTree iterates over the channel and
// inserts each individual post into our RBT,
// setting the post ID as the node's key.
func addPostToTree(tree *tree.RBT, collection <-chan models.Post, group *sync.WaitGroup) {
defer group.Done()
for post := range collection {
// ignore return value & error here:
// we don't care about the returned key and
// error is only ever if a duplicate is attempted to be added -- we don't care
tree.Insert(post.ID, post)
}
}
我可以向端點發出一個請求，但是一旦嘗試提交第二個請求，我的程式就會失敗并顯示panic: sync: negative WaitGroup counter.
我的問題是為什么我的 WaitGroup 計數器會變成負數？我確保添加到等待組并標記我的 goroutine 何時完成。
If the waitgroup is negative on the second request, then that must mean that the first time I allocate a waitgroup and add 4 to it is being skipped... why? Does this have something to do with closing channels, maybe? And if so, where do I close a channel?
Also -- does anyone have tips for debugging goroutines?
Thanks for your help.
uj5u.com熱心網友回復：
首先，整個設計相當復雜。最后提到了我的想法。
您的代碼中有兩個問題：
posts通道永遠不會關閉，因此addPostToTree可能永遠不會存在回圈，導致一個 waitGroup 永遠不會減少（在您的情況下程式掛起）。程式有可能無限期等待并出現死鎖（認為其他 goroutine 會釋放它，但所有 goroutine 都卡住了）。
解決方法：可以關閉postChan通道。但是如何？始終建議制作人始終關閉頻道，但您有多個制作人。所以最好的選擇是，等待所有生產者完成，然后關閉通道。為了等待所有生產者完成，您需要創建另一個 waitGroup 并使用它來跟蹤子例程。
代碼：
// fetch completes a GET request to the endpoint
func fetch(posts chan<- models.Post, tags <-chan string, errs chan<- error, group *sync.WaitGroup) {
postsWG := &sync.WaitGroup{}
for tag := range tags {
ep, err := formURL(tag)
if err != nil {
errs <- err
}
postsWG.Add(1) // QUESTION should I use a separate wait group here?
go func() {
resp, err := http.Get(ep.String())
if err != nil {
errs <- err
}
container := models.PostContainer{}
err = json.NewDecoder(resp.Body).Decode(&container)
defer resp.Body.Close()
go insertPosts(posts, container.Posts, postsWG)
}()
}
defer func() {
postsWG.Wait()
close(posts)
group.Done()
}()
}
- 現在，我們有另一個問題，主 waitGroup 應該用
3而不是初始化4。這是因為主程式只啟動了另外 3 個程式wg.Add(3)，所以它只需要跟蹤那些程式。對于子例程，我們使用了不同的 waitGroup，因此這不再是父例程的頭痛問題。
代碼：
errChan := make(chan error) // for synchronizing errors across goroutines
wg := &sync.WaitGroup{} // for synchronizing goroutines
wg.Add(3)
// create a go func to synchronize our wait groups
// once all goroutines are finished, we can close our errChan
TLDR--
復雜的設計 - 由于主等待組在一個地方啟動，但是每個 goroutine 都在根據需要修改這個 waitGroup。所以這沒有單一的所有者，這使得除錯和維護變得超級復雜（ 不能確保它不會有錯誤）。
我建議將其分解并為每個子例程設定單獨的跟蹤器。這樣，正在啟動更多例程的呼叫者只能專注于跟蹤其子 goroutine。然后，此例程將僅在完成后通知其父級 waitGroup（并且其子級完成后，而不是讓子級例程直接通知祖父母）。
Also, in fetch method after making the HTTP call and getting the response, why create another goroutine to process this data? Either way this goroutine cannot exit until the data insertion happens, nor is it doing someother action which data processing happens. From what I understand, The second goroutine is redundant.
group.Add(1) // QUESTION should I add a separate wait group here and pass it to insertPosts?
go insertPosts(posts, container.Posts, group)
defer group.Done()
轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/352548.html
標籤：go concurrency channel goroutine