控制Goroutine并发量的解决方案 Go控制协裎并发数量的用法及实际中的一个案 ...

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

控制Goroutine并发量的解决方案 Go控制协裎并发数量的用法及实际中的一个案 ...

原作者: [db:作者] 来自: [db:来源] 收藏邀请

前言

　　Go语言虽然开并发Goroutine特别简单，但是实际中如果不控制并发的数量会导致资源的浪费以及同时占用大量服务资源（http连接、数据库连接、文件句柄等）导致服务性能下降！

　　笔者之前总结过一篇在业务代码中控制并发数量的文章：Go控制协裎并发数量的用法及实际中的一个案例

ants库实现链接池的效果控制并发量

　　今天介绍另外一个控制并发数量的第三方库：ants

　　简而言之，ants库通过实现“Goroutine链接池”来限制Goroutine的数量：通过NewPool函数创建一个goroutine pool实现具体效果。

　　创建完 goroutine pool 后，通过 pool.Submit 方法向 pool 中提交任务。
　　如果 pool 中尚有空闲的 goroutine worker，则 pool.Submit 立即返回；否则根据 pool 的配置，pool.Submit 立即返回错误或等待有空闲 goroutine worker 成功接收任务后返回。

使用案例

　　使用之前记当然是 go get一下：

go get github.com/panjf2000/ants

基本使用

　　最基本的使用场景是：提交任务，等待任务完成并获取结果。

package test1

import (
    "fmt"
    "github.com/panjf2000/ants/v2"
    "sync"
    "testing"
)

func sum(a, b int) int {
    return a + b
}

func wrapSum(i int, ch chan int, wg *sync.WaitGroup) func() {
    return func() {
        defer wg.Done()
        ch <- sum(i, i)
    }
}

func TestT1(t *testing.T) {

    var wg sync.WaitGroup
    ch := make(chan int, 10)
    // ants.Release 相当于调用 defaultPool.Release,停止 defaultPool 中所有的 goroutine worker.
    defer ants.Release()
    for i := 0; i < 10; i++ {
        wg.Add(1)
        // ants.Submit 相当于调用 defaultPool.Submit，而 defaultPool 是在 package 初始化时 ants 库创建的
        if err := ants.Submit(wrapSum(i, ch, &wg)); err != nil {
            return
        }
    }
    wg.Wait()
    close(ch)
    for v := range ch {
        fmt.Println(v)
    }

}

g1_test.go

　　需要注意以下几点：

　　1、这里使用的是ants包默认的链接池（ants.Submit方法），打开源码可以看到链接池的容量大小为: math.MaxInt32*（2147483647），所以实际中推荐大家自己控制链接池的容量大小。

　　2、Submit 方法只接受 func() 类型的参数，如果提交的任务有参数，需要自己 wrap。

　　3、ants 没有提供返回值机制，任务的执行结果需要自己进行处理，例子中用了一个带 buffer 的 channel。需要注意的是，当 pool 中有多个任务时，任务的返回值不是根据任务的提交顺序进行排序的，任务的返回顺序取决于调用时机，可以认为是随机的。

　　4、ants 没有提供等待所有任务完成的机制，例子中用了 sync.WaitGroup 实现了等待所有任务完成的机制，否则 main goroutine 可能会在任务执行结束前退出。

配置pool为nonblocking状态的情况

　　以下示例将 pool 配置为 nonblocking。在这种情况下，当 pool 中没有可用的 goroutine worker 时，Submit 会直接返回错误 ants.ErrPoolOverload，而不会等待提交任务成功才返回。

　　另外这里可以配置链接池的大小（ants.NewPool方法）:

package test1

import (
    "fmt"
    "github.com/panjf2000/ants/v2"
    "testing"
)

func hangForever() {
    ch := make(chan int)
    ch <- 10
}

func TestT2(t *testing.T) {
    pool, err := ants.NewPool(10, ants.WithNonblocking(true))
    if err != nil {
        return
    }
    defer pool.Release()
    for i := 0; i < 10; i++ {
        if err := pool.Submit(hangForever); err != nil {
            return
        }
    }
    if err := pool.Submit(func() { fmt.Println("hello") }); err != nil {
        fmt.Printf("err=ErrPoolOverload:%t\n", err == ants.ErrPoolOverload)
    }
}

g2_test.go

关于超时任务的处理 ***

　　在实际中我们往往会希望在摸一个Goroutine执行任务超时或者其他一些情况下退出而不是一直占用着资源！

　　但是由于线程才是操作系统可调度的最小的单位，Goroutine是代码级别的并发，由于GMP模型的限制，我们并不能确定开启的子Goroutine什么时候执行，Go中也没有像epoll那样的“轮询机制”——专门开一个协程去轮询其他的子Goroutine管理它们，所以想要真正的去实现子Goroutine的超时退出需要程序员们在业务代码中做相应的逻辑处理。
　　我这里使用context去简单处理超时的Goroutine：

package test1

import (
    "context"
    "fmt"
    "github.com/panjf2000/ants/v2"
    "testing"
    "time"
)

func expensiveTask2(ctx context.Context, a, b int) (int, error) {
    select {
    // simulate an expensive task
    case <-time.After(10 * time.Second):
        return a + b, nil
    case <-ctx.Done():
        return 0, ctx.Err()
    }
}

func wrap2(ctx context.Context) func() {
    return func() {
        sum, err := expensiveTask2(ctx, 10, 20)
        if err != nil {
            fmt.Printf("error is %v\n", err)
        } else {
            fmt.Printf("sum is %d\n", sum)
        }
    }
}

func TestT31(t *testing.T) {
    pool, err := ants.NewPool(1)
    if err != nil {
        return
    }
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    if err := pool.Submit(wrap2(ctx)); err != nil {
        return
    }
    // wait other goroutines.
    for i := 0; i < 20; i++ {
        time.Sleep(time.Second)
        fmt.Printf("main waits for %d seconds\n", i+1)
    }
}