go - 如何确保在 goroutines 中启动的 goroutines 彼此同步?
问题描述
这是我第一次使用 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 何时完成。
如果 waitgroup 在第二个请求上为负数,那么这一定意味着我第一次分配一个 waitgroup 并添加 4 被跳过......为什么?这可能与关闭频道有关吗?如果是这样,我在哪里关闭频道?
另外——有没有人有调试 goroutines 的提示?
谢谢你的帮助。
解决方案
首先,整个设计相当复杂。说到最后我的想法。
您的代码中有两个问题:
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(及其子程序完成,而不是让子程序直接通知祖父母)。
另外,fetch
在进行 HTTP 调用并获得响应后的方法中,为什么要创建另一个 goroutine 来处理这些数据?无论哪种方式,这个 goroutine 在数据插入发生之前都无法退出,也不会执行数据处理发生的其他操作。据我了解,第二个 goroutine 是多余的。
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()
推荐阅读
- typescript - 使用模块 xlsx-style 和 xlsx 将图像添加到 xlsx 文档
- reactjs - 我无法将变量传递给函数 ReactJS 之外的函数
- java - 合并多个 LiveData 源?
- pandas - Pandas:按整数选择数据框中的多列
- c# - 当方法被另一个方法调用时,RabbitMQ Consumer 不检索消息
- javascript - 我想为我的 Quiz React 应用程序启动倒数计时器(小时:分钟:秒),如何仅在选择主题时启动计时器?
- java - 在 JPA 一对多关系中分配给 NULL 的外键
- python - ModuleNotFoundError 即使在 Google Colab 中导入之后
- python - 在 C++ 中使用 Mitsuba2 插件的示例
- python - Pandas:为列中的下 n 个单元格写入相同的值