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 
