Go 每日一库之 ants

简介

处理大量并发是 Go 语言的一大优势。语言内置了方便的并发语法，可以非常方便的创建很多个轻量级的 goroutine 并发处理任务。相比于创建多个线程，goroutine 更轻量、资源占用更少、切换速度更快、无线程上下文切换开销更少。但是受限于资源总量，系统中能够创建的 goroutine 数量也是受限的。默认每个 goroutine 占用 8KB 内存，一台 8GB 内存的机器满打满算也只能创建 8GB/8KB = 1000000 个 goroutine，更何况系统还需要保留一部分内存运行日常管理任务，go 运行时需要内存运行 gc、处理 goroutine 切换等。使用的内存超过机器内存容量，系统会使用交换区（swap），导致性能急速下降。我们可以简单验证一下创建过多 goroutine 会发生什么：

func main() {
  var wg sync.WaitGroup
  wg.Add(10000000)
  for i := 0; i < 10000000; i++ {
    go func() {
      time.Sleep(1 * time.Minute)
    }()
  }
  wg.Wait()
}

在我的机器上（8G内存）运行上面的程序会报errno 1455，即Out of Memory错误，这很好理解。谨慎运行。

另一方面，goroutine 的管理也是一个问题。goroutine 只能自己运行结束，外部没有任何手段可以强制j结束一个 goroutine。如果一个 goroutine 因为某种原因没有自行结束，就会出现 goroutine 泄露。此外，频繁创建 goroutine 也是一个开销。

鉴于上述原因，自然出现了与线程池一样的需求，即 goroutine 池。一般的 goroutine 池自动管理 goroutine 的生命周期，可以按需创建，动态缩容。向 goroutine 池提交一个任务，goroutine 池会自动安排某个 goroutine 来处理。

ants就是其中一个实现 goroutine 池的库。

快速使用

本文代码使用 Go Modules。

创建目录并初始化：

$ mkdir ants && cd ants
$ go mod init github.com/darjun/go-daily-lib/ants

安装ants库，使用v2版本：

$ go get -u github.com/panjf2000/ants/v2

我们接下来要实现一个计算大量整数和的程序。首先创建基础的任务结构，并实现其执行任务方法：

type Task struct {
  index int
  nums  []int
  sum   int
  wg    *sync.WaitGroup
}

func (t *Task) Do() {
  for _, num := range t.nums {
    t.sum += num
  }

  t.wg.Done()
}

很简单，就是将一个切片中的所有整数相加。

然后我们创建 goroutine 池，注意池使用完后需要手动关闭，这里使用defer关闭：

p, _ := ants.NewPoolWithFunc(10, taskFunc)
defer p.Release()

func taskFunc(data interface{}) {
  task := data.(*Task)
  task.Do()
  fmt.Printf("task:%d sum:%d\n", task.index, task.sum)
}

上面调用了ants.NewPoolWithFunc()创建了一个 goroutine 池。第一个参数是池容量，即池中最多有 10 个 goroutine。第二个参数为每次执行任务的函数。当我们调用p.Invoke(data)的时候，ants池会在其管理的 goroutine 中找出一个空闲的，让它执行函数taskFunc，并将data作为参数。

接着，我们模拟数据，做数据切分，生成任务，交给 ants 处理：

const (
  DataSize    = 10000
  DataPerTask = 100
)

nums := make([]int, DataSize, DataSize)
for i := range nums {
  nums[i] = rand.Intn(1000)
}

var wg sync.WaitGroup
wg.Add(DataSize / DataPerTask)
tasks := make([]*Task, 0, DataSize/DataPerTask)
for i := 0; i < DataSize/DataPerTask; i++ {
  task := &Task{
    index: i + 1,
    nums:  nums[i*DataPerTask : (i+1)*DataPerTask],
    wg:    &wg,
  }

  tasks = append(tasks, task)
  p.Invoke(task)
}

wg.Wait()
fmt.Printf("running goroutines: %d\n", ants.Running())

随机生成 10000 个整数，将这些整数分为 100 份，每份 100 个，生成Task结构，调用p.Invoke(task)处理。wg.Wait()等待处理完成，然后输出ants正在运行的 goroutine 数量，这时应该是 0。

最后我们将结果汇总，并验证一下结果，与直接相加得到的结果做一个比较：

var sum int
for _, task := range tasks {
  sum += task.sum
}

var expect int
for _, num := range nums {
  expect += num
}

fmt.Printf("finish all tasks, result is %d expect:%d\n", sum, expect)

运行：

$ go run main.go
...
task:96 sum:53275
task:88 sum:50090
task:62 sum:57114
task:45 sum:48041
task:82 sum:45269
running goroutines: 0
finish all tasks, result is 5010172 expect:5010172

确实，任务完成之后，正在运行的 goroutine 数量变为 0。而且我们验证了，结果没有偏差。另外需要注意，goroutine 池中任务的执行顺序是随机的，与提交任务的先后没有关系。由上面运行打印的任务标识我们也能发现这一点。

函数作为任务

ants支持将一个不接受任何参数的函数作为任务提交给 goroutine 运行。由于不接受参数，我们提交的函数要么不需要外部数据，只需要处理自身逻辑，否则就必须用某种方式将需要的数据传递进去，例如闭包。

提交函数作为任务的 goroutine 池使用ants.NewPool()创建，它只接受一个参数表示池子的容量。调用池子对象的Submit()方法来提交任务，将一个不接受任何参数的函数传入。

最开始的例子可以改写一下。增加一个任务包装函数，将任务需要的参数作为包装函数的参数。包装函数返回实际的任务函数，该任务函数就可以通过闭包访问它需要的数据了：

type taskFunc func()

func taskFuncWrapper(nums []int, i int, sum *int, wg *sync.WaitGroup) taskFunc {
  return func() {
    for _, num := range nums[i*DataPerTask : (i+1)*DataPerTask] {
      *sum += num
    }

    fmt.Printf("task:%d sum:%d\n", i+1, *sum)
    wg.Done()
  }
}

调用ants.NewPool(10)创建 goroutine 池，同样池子用完需要释放，这里使用defer：

p, _ := ants.NewPool(10)
defer p.Release()

生成模拟数据，切分任务。提交任务给ants池执行，这里使用taskFuncWrapper()包装函数生成具体的任务，然后调用p.Submit()提交：

nums := make([]int, DataSize, DataSize)
for i := range nums {
  nums[i] = rand.Intn(1000)
}

var wg sync.WaitGroup
wg.Add(DataSize / DataPerTask)
partSums := make([]int, DataSize/DataPerTask, DataSize/DataPerTask)
for i := 0; i < DataSize/DataPerTask; i++ {
  p.Submit(taskFuncWrapper(nums, i, &partSums[i], &wg))
}
wg.Wait()

汇总结果，验证：

var sum int
for _, partSum := range partSums {
  sum += partSum
}

var expect int
for _, num := range nums {
  expect += num
}
fmt.Printf("running goroutines: %d\n", ants.Running())
fmt.Printf("finish all tasks, result is %d expect is %d\n", sum, expect)

这个程序的功能与最开始的完全相同。

执行流程

GitHub 仓库中有个执行流程图，我重新绘制了一下：

[图片上传失败...(image-3b5c14-1623109126905)]

执行流程如下：

  • 初始化 goroutine 池；
  • 提交任务给 goroutine 池，检查是否有空闲的 goroutine：
    • 有，获取空闲 goroutine
    • 无，检查池中的 goroutine 数量是否已到池容量上限：
      • 已到上限，检查 goroutine 池是否是非阻塞的：
        • 非阻塞，直接返回nil表示执行失败
        • 阻塞，等待 goroutine 空闲
      • 未到上限，创建一个新的 goroutine 处理任务
  • 任务处理完成，将 goroutine 交还给池，以待处理下一个任务

选项

ants提供了一些选项可以定制 goroutine 池的行为。选项使用Options结构定义：

// src/github.com/panjf2000/ants/options.go
type Options struct {
  ExpiryDuration time.Duration
  PreAlloc bool
  MaxBlockingTasks int
  Nonblocking bool
  PanicHandler func(interface{})
  Logger Logger
}

各个选项含义如下：

  • ExpiryDuration：过期时间。表示 goroutine 空闲多长时间之后会被ants池回收
  • PreAlloc：预分配。调用NewPool()/NewPoolWithFunc()之后预分配worker（管理一个工作 goroutine 的结构体）切片。而且使用预分配与否会直接影响池中管理worker的结构。见下面源码
  • MaxBlockingTasks：最大阻塞任务数量。即池中 goroutine 数量已到池容量，且所有 goroutine 都处理繁忙状态，这时到来的任务会在阻塞列表等待。这个选项设置的是列表的最大长度。阻塞的任务数量达到这个值后，后续任务提交直接返回失败
  • Nonblocking：池是否阻塞，默认阻塞。提交任务时，如果ants池中 goroutine 已到上限且全部繁忙，阻塞的池会将任务添加的阻塞列表等待（当然受限于阻塞列表长度，见上一个选项）。非阻塞的池直接返回失败
  • PanicHandler：panic 处理。遇到 panic 会调用这里设置的处理函数
  • Logger：指定日志记录器

NewPool()部分源码：

if p.options.PreAlloc {
  if size == -1 {
    return nil, ErrInvalidPreAllocSize
  }
  p.workers = newWorkerArray(loopQueueType, size)
} else {
  p.workers = newWorkerArray(stackType, 0)
}

使用预分配时，创建loopQueueType类型的结构，反之创建stackType类型。这是ants定义的两种管理worker的数据结构。

ants定义了一些With*函数来设置这些选项：

func WithOptions(options Options) Option {
  return func(opts *Options) {
    *opts = options
  }
}

func WithExpiryDuration(expiryDuration time.Duration) Option {
  return func(opts *Options) {
    opts.ExpiryDuration = expiryDuration
  }
}

func WithPreAlloc(preAlloc bool) Option {
  return func(opts *Options) {
    opts.PreAlloc = preAlloc
  }
}

func WithMaxBlockingTasks(maxBlockingTasks int) Option {
  return func(opts *Options) {
    opts.MaxBlockingTasks = maxBlockingTasks
  }
}

func WithNonblocking(nonblocking bool) Option {
  return func(opts *Options) {
    opts.Nonblocking = nonblocking
  }
}

func WithPanicHandler(panicHandler func(interface{})) Option {
  return func(opts *Options) {
    opts.PanicHandler = panicHandler
  }
}

func WithLogger(logger Logger) Option {
  return func(opts *Options) {
    opts.Logger = logger
  }
}

这里使用了 Go 语言中非常常见的一种模式，我称之为选项模式，非常方便地构造有大量参数，且大部分有默认值或一般不需要显式设置的对象。

我们来验证几个选项。

最大等待队列长度

ants池设置容量之后，如果所有的 goroutine 都在处理任务。这时提交的任务默认会进入等待队列，WithMaxBlockingTasks(maxBlockingTasks int)可以设置等待队列的最大长度。超过这个长度，提交任务直接返回错误：

func wrapper(i int, wg *sync.WaitGroup) func() {
  return func() {
    fmt.Printf("hello from task:%d\n", i)
    time.Sleep(1 * time.Second)
    wg.Done()
  }
}

func main() {
  p, _ := ants.NewPool(4, ants.WithMaxBlockingTasks(2))
  defer p.Release()

  var wg sync.WaitGroup
  wg.Add(8)
  for i := 1; i <= 8; i++ {
    go func(i int) {
      err := p.Submit(wrapper(i, &wg))
      if err != nil {
        fmt.Printf("task:%d err:%v\n", i, err)
        wg.Done()
      }
    }(i)
  }

  wg.Wait()
}

上面代码中，我们设置 goroutine 池的容量为 4，最大阻塞队列长度为 2。然后一个 for 提交 8 个任务，期望结果是：4 个任务在执行，2 个任务在等待，2 个任务提交失败。运行结果：

hello from task:8
hello from task:5
hello from task:4
hello from task:6
task:7 err:too many goroutines blocked on submit or Nonblocking is set
task:3 err:too many goroutines blocked on submit or Nonblocking is set
hello from task:1
hello from task:2

我们看到提交任务失败，打印too many goroutines blocked ...。

代码中有 4 点需要注意：

  • 提交任务必须并行进行。如果是串行提交，第 5 个任务提交时由于池中没有空闲的 goroutine 处理该任务，Submit()方法会被阻塞，后续任务就都不能提交了。也就达不到验证的目的了
  • 由于任务可能提交失败，失败的任务不会实际执行，所以实际上wg.Done()次数会小于 8。因而在err != nil分支中我们需要调用一次wg.Done()。否则wg.Wait()会永远阻塞
  • 为了避免任务执行过快，空出了 goroutine，观察不到现象，每个任务中我使用time.Sleep(1 * time.Second)休眠 1s
  • 由于 goroutine 之间的执行顺序未显式同步，故每次执行的顺序不确定

由于简单起见，前面的例子中Submit()方法的返回值都被我们忽略了。实际开发中一定不要忽略。

非阻塞

ants池默认是阻塞的，我们可以使用WithNonblocking(nonblocking bool)设置其为非阻塞。非阻塞的ants池中，在所有 goroutine 都在处理任务时，提交新任务会直接返回错误：

func main() {
  p, _ := ants.NewPool(2, ants.WithNonblocking(true))
  defer p.Release()

  var wg sync.WaitGroup
  wg.Add(3)
  for i := 1; i <= 3; i++ {
    err := p.Submit(wrapper(i, &wg))
    if err != nil {
      fmt.Printf("task:%d err:%v\n", i, err)
      wg.Done()
    }
  }

  wg.Wait()
}

使用上个例子中的wrapper()函数，ants池容量设置为 2。连续提交 3 个任务，期望结果前两个任务正常执行，第 3 个任务提交时返回错误：

hello from task:2
task:3 err:too many goroutines blocked on submit or Nonblocking is set
hello from task:1

panic 处理器

一个鲁棒性强的库一定不会忽视错误的处理，特别是宕机相关的错误。在 Go 语言中就是 panic，也被称为运行时恐慌，在程序运行的过程中产生的严重性错误，例如索引越界，空指针解引用等，都会触发 panic。如果不处理 panic，程序会直接意外退出，可能造成数据丢失的严重后果。

ants中如果 goroutine 在执行任务时发生panic，会终止当前任务的执行，将发生错误的堆栈输出到os.Stderr。注意，该 goroutine 还是会被放回池中，下次可以取出执行新的任务。

func wrapper(i int, wg *sync.WaitGroup) func() {
  return func() {
    fmt.Printf("hello from task:%d\n", i)
    if i%2 == 0 {
      panic(fmt.Sprintf("panic from task:%d", i))
    }
    wg.Done()
  }
}

func main() {
  p, _ := ants.NewPool(2)
  defer p.Release()

  var wg sync.WaitGroup
  wg.Add(3)
  for i := 1; i <= 2; i++ {
    p.Submit(wrapper(i, &wg))
  }

  time.Sleep(1 * time.Second)
  p.Submit(wrapper(3, &wg))
  p.Submit(wrapper(5, &wg))
  wg.Wait()
}

我们让偶数个任务触发panic。提交两个任务，第二个任务一定会触发panic。触发panic之后，我们还可以继续提交任务 3、5。注意这里没有 4，提交任务 4 还是会触发panic。

上面的程序需要注意 2 点：

  • 任务函数中wg.Done()是在panic方法之后，如果触发了panic，函数中的其他正常逻辑就不会再继续执行了。所以我们虽然wg.Add(3)，但是一共提交了 4 个任务，其中一个任务触发了panic，wg.Done()没有正确执行。实际开发中，我们一般使用defer语句来确保wg.Done()一定会执行
  • 在 for 循环之后，我添加了一行代码time.Sleep(1 * time.Second)。如果没有这一行，后续的两条Submit()方法可以直接执行，可能会导致任务很快就完成了，wg.Wait()直接返回了，这时panic的堆栈还没有输出。你可以尝试注释掉这行代码运行看看结果

除了ants提供的默认 panic 处理器，我们还可以使用WithPanicHandler(paincHandler func(interface{}))指定我们自己编写的 panic 处理器。处理器的参数就是传给panic的值：

func panicHandler(err interface{}) {
  fmt.Fprintln(os.Stderr, err)
}

p, _ := ants.NewPool(2, ants.WithPanicHandler(panicHandler))
defer p.Release()

其余代码与上面的完全相同，指定了panicHandler后触发panic就会执行它。运行：

hello from task:2
panic from task:2
hello from task:1
hello from task:5
hello from task:3

看到输出了传给panic函数的字符串（第二行输出）。

默认池

为了方便使用，很多 Go 库都喜欢提供其核心功能类型的一个默认实现。可以直接通过库提供的接口调用。例如net/http，例如ants。ants库中定义了一个默认的池，默认容量为MaxInt32。goroutine 池的各个方法都可以直接通过ants包直接访问：

// src/github.com/panjf2000/ants/ants.go
defaultAntsPool, _ = NewPool(DefaultAntsPoolSize)

func Submit(task func()) error {
  return defaultAntsPool.Submit(task)
}

func Running() int {
  return defaultAntsPool.Running()
}

func Cap() int {
  return defaultAntsPool.Cap()
}

func Free() int {
  return defaultAntsPool.Free()
}

func Release() {
  defaultAntsPool.Release()
}

func Reboot() {
  defaultAntsPool.Reboot()
}

直接使用：

func main() {
  defer ants.Release()

  var wg sync.WaitGroup
  wg.Add(2)
  for i := 1; i <= 2; i++ {
    ants.Submit(wrapper(i, &wg))
  }
  wg.Wait()
}

默认池也需要Release()。

总结

本文介绍了 goroutine 池的由来，并借由ants库介绍了基本的使用方法，和一些细节。ants源码不多，去掉测试的核心代码只有 1k 行左右，建议有时间、感兴趣的童鞋深入阅读。

大家如果发现好玩、好用的 Go 语言库，欢迎到 Go 每日一库 GitHub 上提交 issue😄

参考

  1. ants GitHub：github.com/panjf2000/ants
  2. Go 每日一库 GitHub：https://github.com/darjun/go-daily-lib

我

我的博客：https://darjun.github.io

欢迎关注我的微信公众号【GoUpUp】，共同学习，一起进步~

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 210,914评论 6赞 490
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 89,935评论 2赞 383
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 156,531评论 0赞 345
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,309评论 1赞 282
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,381评论 5赞 384
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,730评论 1赞 289
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,882评论 3赞 404
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,643评论 0赞 266
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,095评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,448评论 2赞 325
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,566评论 1赞 339
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,253评论 4赞 328
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,829评论 3赞 312
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,715评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,945评论 1赞 264
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,248评论 2赞 360
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,440评论 2赞 348

推荐阅读更多精彩内容

  • go并发编程放弃
    go并发编程入门到放弃 并发和并行 并发:一个处理器同时处理多个任务。 并行:多个处理器或者是多核的处理器同时处理...
    yangyunfeng阅读 555评论 0赞 2
  • Go并发
    Go语言中的并发编程 并发是编程里面一个非常重要的概念，Go语言在语言层面天生支持并发，这也是Go语言流行的一个很...
    吴佳浩阅读 355评论 0赞 1
  • Go 并发编程
    1 并发与并行 Erlang之父Joe Armstrong曾经以下图解释并发与并行。 并发在图中的解释是两队人排队...
    泥人冷风阅读 212评论 0赞 0
  • Go实现一个协程池
    转载自:超详细的讲解Go中如何实现一个协程池 并发（并行），一直以来都是一个编程语言里的核心主题之一，也是被开发者...
    紫云02阅读 1,028评论 0赞 1
  • 张艺谋要在电影院开“画展”，预告片孙俪都是戏，关晓彤可期待！
    今天青石的票圈出镜率最高的，莫过于张艺谋的新片终于定档了。 一张满溢着水墨风的海报一次次的出现在票圈里，也就是老谋...
    青石电影阅读 10,312评论 1赞 2