Golang中Waitgroup的替代库有哪些

Golang中Waitgroup的替代库有哪些 我正在考虑编写一个替代 sync.Waitgroup 的库。 我目前觉得缺失或不喜欢的地方：

  1. 缺少获取等待组长度的方法
  2. 当等待组计数器为零时调用 waitgroup.Done() 会报错。
  3. 当等待组为 nil 时调用 waitgroup.Done() 和 waitgroup.Add() 会引发空指针错误。

有时我需要使用一些函数/goroutine，它们可能使用也可能不使用等待组。

例如：

type MyStruct sruct{
wg *sync.Waitgroup
}

func (m *MyStruct) DoSomething(){
wg.Add()
defer wg.Done()
.
.
}

有什么建议或类似的包吗？


更多关于Golang中Waitgroup的替代库有哪些的实战教程也可以访问 https://www.itying.com/category-94-b0.html

5 回复

我认为你需要重构你的代码。看看这个：https://play.golang.org/p/0HcRPYIiWF7。现在你的 MyStruct 不需要知道或关心是否存在 WaitGroup。WaitGroup 仅在需要控制同步的那个函数中使用。

更多关于Golang中Waitgroup的替代库有哪些的实战系列教程也可以访问 https://www.itying.com/category-94-b0.html


第一点可能有些价值。但我看不出为什么第2点和第3点需要改变。

skillian:

func (m *MyStruct) DoSomething() {
    if m.wg != nil {
        m.wg.Add(1)
        defer m.wg.Done()
    }
    // ...
}

是的，这个方法可行，但我需要在每个函数中都写这段代码。在主代码中并不需要使用 waitgroup，只有在测试中才需要。

这段代码不会等待，因为程序在 goroutine 初始化之前就结束了：Go Playground - The Go Programming Language

这段代码会按预期运行：Go Playground - The Go Programming Language

我只是不满意手动管理 wg.Add 和 wg.Done。wg.Add 必须在启动 goroutine 之前调用，否则程序会直接执行后续代码并阻塞，因为 goroutine 可能没来得及在 wg.Wait() 执行前运行 wg.Add(1)。

WaitGroup 的文档中写道：

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

你是在这样使用它吗？还是你在解决一个不同的问题，因此需要一个不同的解决方案？

method for getting length of waitgroup

你获取这个信息是为了什么？在获取长度之后，另一个 goroutine 可能已经调用了 Done，那么你获取的长度值就无效了。

when using waitgroup.Done() when the waitgroup counter is zero I get an error.

这听起来像是你的代码中存在一个错误。使用 WaitGroup 的正确方式是：用你要运行的 goroutine 数量来初始化它，然后在每个 goroutine 完成时调用 Done。你描述的情况表明，你“告诉” WaitGroup 等待（例如）5 个 goroutine，但随后有 6 个或更多的 goroutine 调用了 Done。

when using waitgroup.Done() and waitgroup.Add() and waitgroup is nil I get nil error.

可以这么说，大多数类型的大多数方法在 nil 指针上调用时都会引发 panic。那么下面这样有什么问题呢？

func (m *MyStruct) DoSomething() {
    if m.wg != nil {
        m.wg.Add(1)
        defer m.wg.Done()
    }
    // ...
}

话虽如此，在 DoSomething 中使用 WaitGroup 的方式是错误的。你不能在调用 Add 的同一个 goroutine 中 defer 调用 Done。你必须在 DoSomething 中稍后启动的那个 goroutine 中 defer 调用 Done：

func (m *MyStruct) DoSomething() {
    if m.wg != nil {
        m.wg.Add(1)
    }
    // ...
    go func() {
        defer m.wg.Done()
        // ...
    }()
    // ...
}

在Go语言中，确实有一些库提供了比标准库sync.WaitGroup更丰富的功能。以下是几个值得考虑的替代方案：

1. errgroup

golang.org/x/sync/errgroup 提供了错误传播和上下文取消功能。

import "golang.org/x/sync/errgroup"

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    
    // 可以获取goroutine数量（通过自定义计数）
    var count int32
    
    for i := 0; i < 5; i++ {
        g.Go(func() error {
            atomic.AddInt32(&count, 1)
            defer atomic.AddInt32(&count, -1)
            
            // 工作逻辑
            return nil
        })
    }
    
    // 等待所有goroutine完成，并处理错误
    if err := g.Wait(); err != nil {
        log.Fatal(err)
    }
}

2. conc

github.com/sourcegraph/conc 提供了更安全的并发原语。

import "github.com/sourcegraph/conc"

func main() {
    var wg conc.WaitGroup
    
    // 安全的Add方法，不会在零值时panic
    wg.Go(func() {
        // 工作逻辑
    })
    
    // 可以获取当前运行的goroutine数量
    // wg.Running() // 需要查看具体API
    
    wg.Wait()
}

3. 自定义实现

你可以自己实现一个更安全的WaitGroup：

type SafeWaitGroup struct {
    sync.WaitGroup
    mu    sync.RWMutex
    count int
}

func (swg *SafeWaitGroup) Add(delta int) {
    swg.mu.Lock()
    defer swg.mu.Unlock()
    
    if swg.count+delta < 0 {
        panic("sync: negative WaitGroup counter")
    }
    swg.count += delta
    swg.WaitGroup.Add(delta)
}

func (swg *SafeWaitGroup) Done() {
    swg.Add(-1)
}

func (swg *SafeWaitGroup) Count() int {
    swg.mu.RLock()
    defer swg.mu.RUnlock()
    return swg.count
}

func (swg *SafeWaitGroup) TryAdd(delta int) bool {
    swg.mu.Lock()
    defer swg.mu.Unlock()
    
    if swg.count+delta < 0 {
        return false
    }
    swg.count += delta
    swg.WaitGroup.Add(delta)
    return true
}

// 使用示例
func main() {
    var swg SafeWaitGroup
    
    // 安全调用
    swg.Add(1)
    go func() {
        defer swg.Done()
        // 工作逻辑
    }()
    
    // 获取当前计数
    fmt.Println("Current count:", swg.Count())
    
    swg.Wait()
}

4. parallel

github.com/koding/parallel 提供了更高级的并行执行控制。

import "github.com/koding/parallel"

func main() {
    p := parallel.NewRun()
    
    // 添加任务
    p.Do(func() error {
        // 工作逻辑
        return nil
    })
    
    // 等待所有任务完成
    if err := p.Wait(); err != nil {
        log.Fatal(err)
    }
}

5. 使用channel的轻量级替代

type TaskGroup struct {
    wg    sync.WaitGroup
    count int32
    mu    sync.RWMutex
}

func (tg *TaskGroup) Run(f func()) {
    if tg == nil {
        f()
        return
    }
    
    tg.mu.Lock()
    tg.wg.Add(1)
    atomic.AddInt32(&tg.count, 1)
    tg.mu.Unlock()
    
    go func() {
        defer func() {
            tg.mu.Lock()
            tg.wg.Done()
            atomic.AddInt32(&tg.count, -1)
            tg.mu.Unlock()
        }()
        f()
    }()
}

func (tg *TaskGroup) Wait() {
    if tg != nil {
        tg.wg.Wait()
    }
}

func (tg *TaskGroup) Count() int {
    if tg == nil {
        return 0
    }
    return int(atomic.LoadInt32(&tg.count))
}

// 使用示例
func main() {
    var tg TaskGroup
    
    tg.Run(func() {
        fmt.Println("Task 1")
    })
    
    tg.Run(func() {
        fmt.Println("Task 2")
    })
    
    fmt.Println("Running tasks:", tg.Count())
    tg.Wait()
}

这些方案都解决了你提到的三个问题：提供了计数获取、安全的Done调用和nil安全处理。errgroup和conc是社区中比较成熟的解决方案，而自定义实现可以完全按照你的需求来设计。

回到顶部