同步缓冲通道和等待组

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

以上就是同步缓冲通道和等待组的详细内容，更多请关注编程网其它相关文章！