14

我需要启动大量的 goroutine 并等待它们的终止。直观的方法似乎是使用一个通道来等到所有这些都完成：

package main

type Object struct {
    //data
}

func (obj *Object) Update(channel chan int) {
    //update data
    channel <- 1
    return
}

func main() {

    channel := make(chan int, n)
    list := make([]Object, n, m)
    for {
        for _, object := range list {
            go object.Update(channel)
        }
        for i := 0; i < n; i++ {
            <-channel
        }
        //now everything has been updated. start again
    }
}

但问题是对象的数量以及 goroutine 的数量可能会发生变化。是否可以更改通道的缓冲区大小？

有没有更优雅的方法来做到这一点？

4

3 回答 3

34

我使用WaitGroup作为解决此问题的方法。翻译您当前的代码，并附上一些日志以明确发生了什么：

package main

import "sync"
import "fmt"
import "time"

type Object struct {
    //data
}

func (obj *Object) Update(wg *sync.WaitGroup) {
    //update data
    time.Sleep(time.Second)
    fmt.Println("Update done")
    wg.Done()
    return
}

func main() {
    var wg sync.WaitGroup
    list := make([]Object, 5)
    for {
        for _, object := range list {
            wg.Add(1)
            go object.Update(&wg)
        }
        //now everything has been updated. start again
        wg.Wait()
        fmt.Println("Group done")
    }
}
于 2013-05-16T15:10:22.633 回答
4

这项任务并非微不足道，编写一个有问题的任务很容易。我建议在 stdlib - 中使用现成的解决方案sync.WaitGroup。引用链接：

WaitGroup 等待一组 goroutine 完成。主 goroutine 调用 Add 来设置要等待的 goroutine 的数量。然后每个 goroutine 运行并在完成时调用 Done。同时，Wait 可以用来阻塞，直到所有的 goroutine 都完成。

于 2013-05-16T15:10:16.023 回答
1

@tjameson 很好地解释了如何使用WaitGroup，如何将WaitGroup对象的引用传递给函数。我要对他的例子做的一个改变是杠杆defer作用Done。我认为这defer ws.Done()应该是您函数中的第一条语句。

我喜欢它WaitGroup的简单。但是，我不喜欢我们需要将引用传递给 goroutine，因为这意味着并发逻辑将与您的业务逻辑混合在一起。

所以我想出了这个通用函数来为我解决这个问题：

// Parallelize parallelizes the function calls
func Parallelize(functions ...func()) {
    var waitGroup sync.WaitGroup
    waitGroup.Add(len(functions))

    defer waitGroup.Wait()

    for _, function := range functions {
        go func(copy func()) {
            defer waitGroup.Done()
            copy()
        }(function)
    }
}

所以你的例子可以这样解决：

type Object struct {
    //data
}

func (obj *Object) Update() {
    //update data
    time.Sleep(time.Second)
    fmt.Println("Update done")
    return
}

func main() {
    functions := []func(){}
    list := make([]Object, 5)
    for _, object := range list {
        function := func(obj Object){ object.Update() }(object)
        functions = append(functions, function)
    }

    Parallelize(functions...)        

    fmt.Println("Group done")
}

如果你想使用它，你可以在这里找到它https://github.com/shomali11/util

于 2017-06-09T01:42:59.093 回答