pond

使用 Go 语言编写的极简且高性能的 goroutine 工作池

动机

这个库旨在提供一种简单的方法来限制对有限资源或服务执行某些函数时的并发性。

一些常见场景包括：

对连接数有限的数据库执行查询
向具有速率/并发限制的 API 发送 HTTP 请求

特性：

零依赖
创建固定大小或动态大小的池
工作 goroutine 仅在需要时创建（检测背压），并在空闲一段时间后自动清除（可配置）
极简 API，用于：
- 创建固定或动态大小的工作池
- 以不管结果的方式向池提交任务
- 向池提交任务并等待完成
- 向池提交带截止期限的任务
- 提交一组任务并等待它们完成
- 提交与 Context 关联的一组任务
- 获取正在运行的工作者（goroutine）数量
- 停止工作池
任务中的 panic 会被优雅处理（可配置 panic 处理程序）
支持非阻塞和阻塞任务提交模式（有缓冲/无缓冲）
在重负载下性能非常高且资源使用效率高，在某些场景下甚至优于无限制的 goroutine（参见基准测试）
可配置的池大小调整策略，有 3 个预设适用于常见场景：积极、平衡和懒惰
完整的池指标，如运行中的工作者数量、队列中等待的任务数等等
新特性（自 v1.7.0 起）：可配置父上下文和带截止期限的优雅关闭
API 参考

如何安装

go get -u github.com/alitto/pond

如何使用

动态大小的工作池

package main

import (
	"fmt"

	"github.com/alitto/pond"
)

func main() {

	// 创建一个可扩展至 100 个工作者的有缓冲（非阻塞）池
	// 缓冲容量为 1000 个任务
	pool := pond.New(100, 1000)

	// 提交 1000 个任务
	for i := 0; i < 1000; i++ {
		n := i
		pool.Submit(func() {
			fmt.Printf("正在运行任务 #%d\n", n)
		})
	}

	// 停止池并等待所有提交的任务完成
	pool.StopAndWait()
}

固定大小的工作池

package main

import (
	"fmt"

	"github.com/alitto/pond"
)

func main() {

	// 创建一个固定工作者数量的无缓冲（阻塞）池
	pool := pond.New(10, 0, pond.MinWorkers(10))

	// 提交 1000 个任务
	for i := 0; i < 1000; i++ {
		n := i
		pool.Submit(func() {
			fmt.Printf("正在运行任务 #%d\n", n)
		})
	}

	// 停止池并等待所有提交的任务完成
	pool.StopAndWait()
}

提交一组任务

package main

import (
	"fmt"

	"github.com/alitto/pond"
)

func main() {

	// 创建一个池
	pool := pond.New(10, 1000)
	defer pool.StopAndWait()

	// 创建一个任务组
	group := pool.Group()

	// 提交一组任务
	for i := 0; i < 20; i++ {
		n := i
		group.Submit(func() {
			fmt.Printf("正在运行组任务 #%d\n", n)
		})
	}

	// 等待组中所有任务完成
	group.Wait()
}

提交与上下文关联的一组任务（自 v1.8.0 起）

此功能为共同任务的子任务提供同步、错误传播和 Context 取消。类似于 golang.org/x/sync/errgroup 包中的 errgroup.Group，但并发性受工作池限制。

package main

import (
	"context"
	"fmt"
	"net/http"

	"github.com/alitto/pond"
)

func main() {

	// 创建一个工作池
	pool := pond.New(10, 1000)
	defer pool.StopAndWait()

	// 创建一个与上下文关联的任务组
	group, ctx := pool.GroupContext(context.Background())

	var urls = []string{
		"https://www.golang.org/",
		"https://www.google.com/",
		"https://www.github.com/",
	}

	// 提交任务以获取每个 URL
	for _, url := range urls {
		url := url
		group.Submit(func() error {
			req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
			resp, err := http.DefaultClient.Do(req)
			if err == nil {
				resp.Body.Close()
			}
			return err
		})
	}

	// 等待所有 HTTP 请求完成
	err := group.Wait()
	if err != nil {
		fmt.Printf("获取 URL 失败：%v", err)
	} else {
		fmt.Println("成功获取所有 URL")
	}
}

池配置选项

MinWorkers

指定任何时候必须运行的最少工作 goroutine 数量。这些 goroutine 在创建池时启动。默认值为 0。示例：

// 这将创建一个有 5 个运行中工作 goroutine 的池
pool := pond.New(10, 1000, pond.MinWorkers(5))

IdleTimeout

定义在从池中移除空闲工作 goroutine 之前等待的时间。默认值为 5 秒。示例：

// 这将创建一个池，在工作者变为空闲 100ms 后移除它们
pool := pond.New(10, 1000, pond.IdleTimeout(100 * time.Millisecond))

PanicHandler

允许配置一个自定义函数来处理提交到池的任务抛出的 panic。默认处理程序只是使用 fmt.Printf 向标准输出写入一条消息，内容如下：Worker exits from a panic: [panic] \n Stack trace: [stack trace]）。示例：

// 自定义 panic 处理函数
panicHandler := func(p interface{}) {
	fmt.Printf("任务发生 panic：%v", p)
}

// 这将创建一个使用自定义 panic 处理程序处理 panic 的池
pool := pond.New(10, 1000, pond.PanicHandler(panicHandler)))

Strategy

配置在检测到背压时用于调整池大小的策略。你可以通过实现 pond.ResizingStrategy 接口创建自定义策略，或选择以下 3 个预设之一：

急切（Eager）：以较高的资源使用为代价最大化响应速度，在某些情况下可能会降低吞吐量。这种策略适用于大部分时间只使用少量容量，偶尔会收到任务突发的工作池。这是默认策略。
平衡（Balanced）：试图在响应速度和吞吐量之间找到平衡。适用于通用工作池或大部分时间运行在50%容量左右的工作池。
懒惰（Lazy）：以响应速度为代价最大化吞吐量。这种策略适用于大部分时间接近最大容量运行的工作池。

// 示例：使用不同的调整策略创建池
eagerPool := pond.New(10, 1000, pond.Strategy(pond.Eager()))
balancedPool := pond.New(10, 1000, pond.Strategy(pond.Balanced()))
lazyPool := pond.New(10, 1000, pond.Strategy(pond.Lazy()))

上下文

为这个池配置一个父上下文，当它被取消时停止所有工作者。默认值为context.Background()。示例：

// 这会创建一个在myCtx被取消时停止的池
pool := pond.New(10, 1000, pond.Context(myCtx))

调整策略

下图说明了随着提交任务数量增加，不同池调整策略的行为。每条线代表池中的工作者goroutine数量（池大小），x轴反映提交的任务数量（累计）。

池调整策略行为

顾名思义，"急切"策略在没有空闲工作者时总是生成一个额外的工作者，这导致池的大小几乎与提交的任务数量呈线性增长。另一方面，"懒惰"策略每N个提交的任务创建一个工作者，其中N是可用CPU的最大数量（GOMAXPROCS）。"平衡"策略代表前两者之间的中间地带，因为它每N/2个提交的任务创建一个工作者。