Golang高效Goroutine管理与取消插件库Pool的使用

包介绍

Pool包实现了一个有限消费者goroutine或无限goroutine池，用于更简单的goroutine处理和取消。

特性

• 使用极其简单，不对您的使用方式做任何假设
• 从返回错误的消费者goroutine自动恢复

Pool v3相比v2的优势

• 对象不再是接口，减少了未来可能出现的破坏性变更
• 现在有两种完全可互换的Pool类型：有限工作池和无限池
• 工作单元使用更简单，现在可以使用work.Wait()而不是<-work.Done

安装

使用go get安装：

go get gopkg.in/go-playground/pool.v3

然后在代码中导入：

import "gopkg.in/go-playground/pool.v3"

重要提示

• 建议从调用函数中取消池或批处理，而不是在工作单元内部取消
• 批处理时不要忘记调用batch.QueueComplete()，否则会导致死锁
• 在阻塞操作后（如等待连接池中的连接）检查WorkUnit.IsCancelled()是您的责任

使用示例

有限池和无限池具有相同的签名并且完全可互换。

单个工作单元示例

package main

import (
	"fmt"
	"time"

	"gopkg.in/go-playground/pool.v3"
)

func main() {
	// 创建限制10个goroutine的池
	p := pool.NewLimited(10)
	defer p.Close()

	// 排队获取用户信息
	user := p.Queue(getUser(13))
	// 排队获取其他信息
	other := p.Queue(getOtherInfo(13))

	// 等待用户信息任务完成
	user.Wait()
	if err := user.Error(); err != nil {
		// 处理错误
	}

	// 使用返回的用户信息
	username := user.Value().(string)
	fmt.Println(username)

	// 等待其他信息任务完成
	other.Wait()
	if err := other.Error(); err != nil {
		// 处理错误
	}

	// 使用返回的其他信息
	otherInfo := other.Value().(string)
	fmt.Println(otherInfo)
}

// 获取用户信息的工作函数
func getUser(id int) pool.WorkFunc {
	return func(wu pool.WorkUnit) (interface{}, error) {
		// 模拟等待TCP连接建立或从连接池获取连接
		time.Sleep(time.Second * 1)

		if wu.IsCancelled() {
			// 返回值不会被使用
			return nil, nil
		}

		// 准备处理...
		return "Joeybloggs", nil
	}
}

// 获取其他信息的工作函数
func getOtherInfo(id int) pool.WorkFunc {
	return func(wu pool.WorkUnit) (interface{}, error) {
		// 模拟等待TCP连接建立或从连接池获取连接
		time.Sleep(time.Second * 1)

		if wu.IsCancelled() {
			// 返回值不会被使用
			return nil, nil
		}

		// 准备处理...
		return "Other Info", nil
	}
}

批处理工作示例

package main

import (
	"fmt"
	"time"

	"gopkg.in/go-playground/pool.v3"
)

func main() {
	// 创建限制10个goroutine的池
	p := pool.NewLimited(10)
	defer p.Close()

	// 创建批处理
	batch := p.Batch()

	// 为了最大速度，在另一个goroutine中排队
	// 这不是必须的，但在所有项目排队完成前不能开始读取结果
	go func() {
		for i := 0; i < 10; i++ {
			batch.Queue(sendEmail("email content"))
		}

		// 不要忘记这个，否则goroutine会死锁
		// 如果调用Cancel()，它会在内部调用QueueComplete()
		batch.QueueComplete()
	}()

	// 处理结果
	for email := range batch.Results() {
		if err := email.Error(); err != nil {
			// 处理错误
			// 可能需要调用batch.Cancel()
		}

		// 使用返回值
		fmt.Println(email.Value().(bool))
	}
}

// 发送邮件的工作函数
func sendEmail(email string) pool.WorkFunc {
	return func(wu pool.WorkUnit) (interface{}, error) {
		// 模拟等待TCP连接建立或从连接池获取连接
		time.Sleep(time.Second * 1)

		if wu.IsCancelled() {
			// 返回值不会被使用
			return nil, nil
		}

		// 准备处理...
		return true, nil // 一切正常，发送nil，如果有错误则发送错误
	}
}

性能基准

在MacBook Pro (Retina, 15-inch, Late 2013) 2.6 GHz Intel Core i7 16 GB 1600 MHz DDR3上使用Go 1.6.2运行测试。

一些基准测试结果：

• BenchmarkLimitedSmallRun 在1.002492008秒内完成了10秒的处理
• BenchmarkLimitedSmallCancel 运行20个作业，在第6个作业取消，运行时间为0秒
• BenchmarkLimitedLargeCancel 运行1000个作业，在第6个作业取消，运行时间为0秒
• BenchmarkLimitedOverconsumeLargeRun 使用25个工作线程运行100个作业，耗时4.027153081秒

许可证

根据MIT许可证分发，更多详情请参见代码中的许可证文件。