golang支持错误处理和并发控制的WaitGroup增强插件go-waitgroup的使用

go-waitgroup介绍

go-waitgroup是一个增强版的WaitGroup包，它允许你使用类似sync.WaitGroup的结构来创建goroutine池并控制并发。

基本用法

使用方式与标准sync.WaitGroup类似，主要区别在于初始化。使用waitgroup.NewWaitGroup时，可以指定并发限制大小。

• 任何大于0的整数都会限制并发goroutine数量
• 如果指定-1或0，所有goroutine将同时运行(和标准sync.WaitGroup一样)

package main

import (
    "fmt"
    "net/http"

    "github.com/pieterclaerhout/go-waitgroup"
)

func main() {
    
    urls := []string{
        "https://www.easyjet.com/",
        "https://www.skyscanner.de/",
        "https://www.ryanair.com",
        "https://wizzair.com/",
        "https://www.swiss.com/",
    }

    // 限制并发数为3
    wg := waitgroup.NewWaitGroup(3)

    for _, url := range urls {
        wg.BlockAdd()  // 阻塞等待直到有可用槽位
        go func(url string) {
            defer wg.Done()
            fmt.Printf("%s: checking\n", url)
            res, err := http.Get(url)
            if err != nil {
                fmt.Println("Error: %v")
            } else {
                defer res.Body.Close()
                fmt.Printf("%s: result: %v\n", url, err)
            }
        }(url)
    }

    wg.Wait()
    fmt.Println("Finished")
}

使用闭包

也可以使用函数闭包使代码更易读：

package main

import (
    "fmt"
    "net/http"

    "github.com/pieterclaerhout/go-waitgroup"
)

func main() {

    urls := []string{
        "https://www.easyjet.com/",
        "https://www.skyscanner.de/",
        "https://www.ryanair.com",
        "https://wizzair.com/",
        "https://www.swiss.com/",
    }

    // 限制并发数为3
    wg := waitgroup.NewWaitGroup(3)

    for _, url := range urls {
        urlToCheck := url
        wg.Add(func() {
            fmt.Printf("%s: checking\n", urlToCheck)
            res, err := http.Get(urlToCheck)
            if err != nil {
                fmt.Println("Error: %v")
            } else {
                defer res.Body.Close()
                fmt.Printf("%s: result: %v\n", urlToCheck, err)
            }
        })
    }

    wg.Wait()
    fmt.Println("Finished")
}

错误处理

如果需要处理错误，可以使用ErrorGroup。它与普通WaitGroup类似但有两点不同：

1. 只能添加返回error的函数
2. 当其中一个任务失败时，其余任务将被取消

package main

import (
    "context"
    "fmt"
    "os"

    "github.com/pieterclaerhout/go-waitgroup"
)

func main() {

    ctx := context.Background()

    // 创建错误组，限制并发数
    wg, ctx := waitgroup.NewErrorGroup(ctx, tc.size)
    if err != nil {
        fmt.Println("Error: %v")
        os.Exit(1)
    }

    // 添加成功任务
    wg.Add(func() error {
        return nil
    })

    // 添加失败任务
    wg.Add(func() error {
        return errors.New("An error occurred")
    })

    if err := wg.Wait(); err != nil {
        fmt.Println("Error: %v")
        os.Exit(1)
    }
}

也可以一次性添加多个函数：

package main

import (
    "context"
    "errors"
    "fmt"
    "os"

    "github.com/pieterclaerhout/go-waitgroup"
)

func main() {

    ctx := context.Background()

    wg, ctx := waitgroup.NewErrorGroup(ctx, tc.size)
    if err != nil {
        fmt.Println("Error: %v")
        os.Exit(1)
    }

    // 批量添加多个函数
    wg.Add(
        func() error {
            return nil
        },
        func() error {
            return errors.New("An error occurred")
        },
    )

    if err := wg.Wait(); err != nil {
        fmt.Println("Error: %v")
        os.Exit(1)
    }
}