同步緩衝通道和等待組

php小編子墨介紹同步緩衝通道和等待群組，這是在並發程式設計中常用的技巧。同步緩衝通道允許多個執行緒之間進行資料傳遞，透過緩衝區來實現執行緒之間的同步。而等待群組則用於管理一組線程，等待某個特定條件滿足後再同時執行。這兩種技術在多執行緒程式設計中，能夠有效解決執行緒間的同步問題，提高程式的並發效能和可靠性。

問題內容

我在使用 waitgroup 與 buffered 通道時遇到問題。問題是 waitgroup 在頻道完全讀取之前關閉，這使得我的頻道讀到一半並在中間中斷。

func main() {
    var wg sync.waitgroup
    var err error

    start := time.now()
    students := make([]studentdetails, 0)
    studentch := make(chan studentdetail, 10000)
    errorch := make(chan error, 1)

    wg.add(1)

    go s.getdetailstudents(rctx, studentch , errorch, &wg, s.link, false)
    go func(ch chan studentdetail, e chan error) {
    
    loop:
        for {
            select {
            case p, ok := <-ch:
                if ok {
                    l.printf("links %s: [%s]\n", p.title, p.link)
                    students = append(students, p)
                } else {
                    l.print("closed channel")
                    break loop
                }
            case err = <-e:
                if err != nil {
                    break
                }
            }
        }
    }(studentch, errorch)
    wg.wait()
    close(studentch)
    close(errorch)
    l.warnln("closed: all wait-groups completed!")
    l.warnf("total items fetched: %d", len(students))

    elapsed := time.since(start)
    l.warnf("operation took %s", elapsed)
}

問題是這個函數是recursive。我的意思是一些 http 呼叫來取得 students，然後根據條件進行更多呼叫。

func (s Student) getDetailStudents(rCtx context.Context, content chan<- studentDetail, errorCh chan<- error, wg *sync.WaitGroup, url string, subSection bool) {
    util.MustNotNil(rCtx)
    L := logger.GetLogger(rCtx)
    defer func() {
        L.Println("Closing all waitgroup!")
        wg.Done()
    }()

    wc := getWC()
    httpClient := wc.Registry.MustHTTPClient()
    res, err := httpClient.Get(url)
    if err != nil {
        L.Fatal(err)
    }
    defer res.Body.Close()
    if res.StatusCode != 200 {
        L.Errorf("status code error: %d %s", res.StatusCode, res.Status)
        errorCh <- errors.New("service_status_code")
        return
    }

    // parse response and return error if found some through errorCh as done above.
    // decide page subSection based on response if it is more.
    if !subSection {
        wg.Add(1)
        go s.getDetailStudents(rCtx, content, errorCh, wg, link, true)
        // L.Warnf("total pages found %d", pageSub.Length()+1)
    }

    // Find students from response list and parse each Student
    students := s.parseStudentItemList(rCtx, item)
    for _, student := range students {
        content <- student
    }
 
    L.Warnf("Calling HTTP Service for %q with total %d record", url, elementsSub.Length())
}

更改變數以避免原始程式碼庫。

問題是一旦等待組完成，學生就會被隨機讀取。我希望保持執行直到所有學生都讀完為止，如果發生錯誤，它應該在遇到錯誤時立即中斷。

解決方法

您需要知道接收 goroutine 何時完成。 waitgroup 為產生 goroutine 執行此操作。因此，您可以使用兩個等待群組：

wg.Add(1)
go s.getDetailStudents(rCtx, studentCh , errorCh, &wg, s.Link, false)
wgReader.Add(1)
go func(ch chan studentDetail, e chan error) {
    defer wgReader.Done()
    ...
}
wg.Wait()
close(studentCh)
close(errorCh)
wgReader.Wait() // Wait for the readers to complete

以上是同步緩衝通道和等待組的詳細內容。更多資訊請關注PHP中文網其他相關文章！