简介
ants是一个高性能的 goroutine 池，实现了对大规模 goroutine 的调度管理、goroutine 复用，允许使用者在开发并发程序的时候限制 goroutine 数量，复用资源，达到更高效执行任务的效果。
功能
- 自动调度海量的 goroutines，复用 goroutines
- 定期清理过期的 goroutines，进一步节省资源
- 提供了大量有用的接口：任务提交、获取运行中的 goroutine 数量、动态调整 Pool 大小、释放 Pool、重启 Pool
- 优雅处理 panic，防止程序崩溃
- 资源复用，极大节省内存使用量；在大规模批量并发任务场景下比原生 goroutine 并发具有更高的性能
- 非阻塞机制
地址
项目中引入地址：
github.com/panjf2000/ants/v2
使用
ants中pool有2种方式NewPool和NewPoolWithFunc这2种方式创建。defaultPool也是NewPool的一种。这两种方式的区别在于，NewPool创建的pool传入的任务是task函数方法，用Submit方法提交一个task任务。而NewPoolWithFunc创建的pool传入的任务是参数，task函数是固定，在NewPoolWithFunc声明的时候已经传入了函数，用Invoke(args interface{})提交参数任务。
因此适用的场景也是不一样的，NewPool适用于多个goroutine处理的事是不一样的(传入不同的函数可以体现出来)，NewPoolWithFunc适用于多个goroutine处理的事是一样的，只不过是要分批处理(函数是固定的，参数是要多次传入的可以体现出来)。
-
文档demo
demo中为我们介绍了defaultPool和NewPoolWithFunc的用法
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/panjf2000/ants/v2"
)
var sum int32
func myFunc(i interface{}) {
n := i.(int32)
atomic.AddInt32(&sum, n)
fmt.Printf("run with %d\n", n)
}
func demoFunc() {
time.Sleep(10 * time.Millisecond)
fmt.Println("Hello World!")
}
func main() {
defer ants.Release()
runTimes := 1000
// Use the common pool.
var wg sync.WaitGroup
syncCalculateSum := func() {
demoFunc()
wg.Done()
}
for i := 0; i < runTimes; i++ {
wg.Add(1)
_ = ants.Submit(syncCalculateSum)
}
wg.Wait()
fmt.Printf("running goroutines: %d\n", ants.Running())
fmt.Printf("finish all tasks.\n")
// Use the pool with a function,
// set 10 to the capacity of goroutine pool and 1 second for expired duration.
p, _ := ants.NewPoolWithFunc(10, func(i interface{}) {
myFunc(i)
wg.Done()
})
defer p.Release()
// Submit tasks one by one.
for i := 0; i < runTimes; i++ {
wg.Add(1)
_ = p.Invoke(int32(i))
}
wg.Wait()
fmt.Printf("running goroutines: %d\n", p.Running())
fmt.Printf("finish all tasks, result is %d\n", sum)
}
- 设置最大等待队列长度
package main
import (
"fmt"
"github.com/panjf2000/ants/v2"
"sync"
"time"
)
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()
}
}
// 阻塞
//我们设置 goroutine 池的容量为 4，最大阻塞队列长度为 2。然后一个 for 提交 8 个任务，期望结果是：4 个任务在执行，2 个任务在等待，2 个任务提交失败。
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 个任务提交失败。
- 非阻塞
package main
import (
"fmt"
"github.com/panjf2000/ants/v2"
"sync"
"time"
)
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()
}
}
// 阻塞
//我们设置 goroutine 池的容量为 4，最大阻塞队列长度为 2。然后一个 for 提交 8 个任务，期望结果是：4 个任务在执行，2 个任务在等待，2 个任务提交失败。
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()
}
ants池容量设置为 2。连续提交 3 个任务，期望结果前两个任务正常执行，第 3 个任务提交时返回错误
- panic 处理器
package main
import (
"fmt"
"github.com/panjf2000/ants/v2"
"sync"
"time"
)
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()
}
}
//我们让偶数个任务触发panic。提交两个任务，第二个任务一定会触发panic。触发panic之后，我们还可以继续提交任务 3、5。注意这里没有 4，提交任务 4 还是会触发panic。
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()
}
- 默认池
package main
import (
"fmt"
"github.com/panjf2000/ants/v2"
"sync"
"time"
)
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() {
defer ants.Release()
var wg sync.WaitGroup
wg.Add(2)
for i := 1; i <= 2; i++ {
ants.Submit(wrapper(i, &wg))
}
wg.Wait()
}
