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中文网其他相关文章!