Golang高性能轻量级goroutine工作池插件pond的使用

pond是一个极简且高性能的Go库，旨在优雅地管理并发任务。它基于Worker Pool模式，允许同时运行大量任务，同时限制同时运行的goroutine数量。

安装

go get -u github.com/alitto/pond/v2

基本使用示例

package main

import (
	"fmt"

	"github.com/alitto/pond/v2"
)

func main() {
	// 创建并发限制为100的工作池
	pool := pond.NewPool(100)

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

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

提交返回错误的任务

// 创建并发限制为100的工作池
pool := pond.NewPool(100)

// 提交返回错误的任务
task := pool.SubmitErr(func() error {
	return errors.New("An error occurred")
})

// 等待任务完成并获取错误
err := task.Wait()

提交返回结果的任务

// 创建接受返回字符串任务的池
pool := pond.NewResultPool[string](10)

// 提交返回字符串的任务
task := pool.Submit(func() (string) {
	return "Hello, World!"
})

// 等待任务完成并获取结果
result, err := task.Wait()
// result = "Hello, World!" 且 err = nil

任务组使用示例

// 创建并发限制为10的工作池
pool := pond.NewPool(10)

// 创建任务组
group := pool.NewGroup()

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

// 等待组内所有任务完成
err := group.Wait()

带上下文的任务组

// 创建并发限制为10的工作池
pool := pond.NewPool(10)

// 创建5秒超时的上下文
timeout, _ := context.WithTimeout(context.Background(), 5*time.Second)

// 创建带上下文的任务组
group := pool.NewGroupContext(timeout)

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

// 等待组内所有任务完成或超时发生，以先到者为准
err := group.Wait()

子池使用示例

// 创建并发限制为10的工作池
pool := pond.NewPool(10)

// 创建子池，使用父池最多5个worker
subpool := pool.NewSubpool(5)

// 向子池提交任务
subpool.Submit(func() {
	fmt.Println("Running task in subpool")
})

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

默认池使用示例

// 向默认池提交任务并等待完成
err := pond.SubmitErr(func() error {
	fmt.Println("Running task in default pool")
	return nil
}).Wait()

if err != nil {
	fmt.Printf("Failed to run task: %v", err)
} else {
	fmt.Println("Task completed successfully")
}

有界任务队列

// 创建队列最多10个任务的池
pool := pond.NewPool(1, pond.WithQueueSize(10))

// 尝试向池提交任务
task, ok := pool.TrySubmit(func() {
	// 执行一些工作
})

// 检查任务是否提交成功
if !ok {
	fmt.Println("Task submission failed because the queue is full")
}

动态调整池大小

// 创建5个worker的池
pool := pond.NewPool(5)

// 提交一些任务
for i := 0; i < 20; i++ {
    pool.Submit(func() {
        // 执行一些工作
    })
}

// 将池大小增加到10个worker
pool.Resize(10)

// 提交更多任务，将使用增加后的容量
for i := 0; i < 20; i++ {
    pool.Submit(func() {
        // 执行一些工作
    })
}

// 将池大小减少回5个worker
pool.Resize(5)

监控指标

pond提供了以下监控指标方法：

• pool.RunningWorkers() int64: 当前运行的worker数量
• pool.SubmittedTasks() uint64: 自池创建以来提交的任务总数
• pool.WaitingTasks() uint64: 当前等待执行的任务数
• pool.SuccessfulTasks() uint64: 成功完成的任务数
• pool.FailedTasks() uint64: 因panic失败的任务数
• pool.CompletedTasks() uint64: 已完成的任务总数(成功+失败)
• pool.DroppedTasks() uint64: 因队列满被丢弃的任务数

pond是一个功能强大且灵活的工作池实现，适用于各种并发场景，特别是需要限制并发量的情况。