2.15 协程池

1. 协程池是什么

复用goroutine的池子

2. 为什么需要协程池

跟如何设计池化技术.md不同，由于 goroutine创建、销毁对象开销小：创建销毁是在用户态， goroutine数量理论上可以无限，一个goroutine占用内存仅仅不到2K 所以99%的情况下不需要协程池 但是在极限情况下，合理复用总是没错的，比如网关这种超高并发低延迟的场景

3. 如何设计协程池

package main

import (
	"fmt"
	"time"
)

type IRunnable interface {
	//执行业务方法
	Run()
}

/*定义一个任务类型Task*/
type Task struct {
	f func() error //一个Task里面有一个具体的业务
}

func (this *Task) Run() {
	this.f()
}

func NewTask(f func() error) *Task {
	return &Task{f: f}
}

type IPool interface {
	//丢进一个任务
	Submit(IRunnable)
}

/*Pool协程池*/
type Pool struct {
	//外部访问这个Channel
	EntryChannel chan IRunnable
	//内部的Task队列
	JobsChannel chan IRunnable
	//最大的worker数量
	workerNum int
}

//启动所有工作协程
func (this *Pool) startWorkers() {
	for i := 0; i < this.workerNum; i++ {
		go this.startOneWorker(i)
	}

	go this.tansferTaskToQueue()
}

//启动一个工作协程
func (this *Pool) startOneWorker(workerId int) {
	for task := range this.JobsChannel {
		task.Run()
		fmt.Println("workerId: ", workerId, " 执行完了一个任务")
	}
}

func (this *Pool) Submit(runnable IRunnable) {
	this.EntryChannel <- runnable
}

//把任务从外部Channel（EntryChannel）转移到内部Channel（JobsChannel）
func (this *Pool) tansferTaskToQueue() {
	for task := range this.EntryChannel {
		this.JobsChannel <- task
	}
}

func NewPool(workerNum int) *Pool {
	pool := &Pool{
		EntryChannel: make(chan IRunnable),
		JobsChannel:  make(chan IRunnable),
		workerNum:    workerNum}
	pool.startWorkers()
	return pool
}

func main() {
	//创建一个任务
	task := NewTask(func() error {
		fmt.Println(time.Now())
		return nil
	})
	//创建4个协程的协程池
	pool := NewPool(4)
	//不停的把任务丢进协程池里
	go func() {
		for true {
			pool.Submit(task)
		}
	}()
	//pool.startWorkers()
	select {}
}

• 简化版

package taskqueue

import (
	"git.code.oa.com/trpc-go/trpc-go/log"
	"git.code.oa.com/trpc-go/trpc-go/metrics"
)

// ITask ...
type ITask interface {
	// 执行业务方法
	Run()
}

// Task ...
type Task struct {
	desc string
	f    func() error
}

// String ...
func (t *Task) String() string {
	return t.desc
}

// NewTask ...
func NewTask(desc string, f func() error) *Task {
	return &Task{
		desc: desc,
		f:    f,
	}
}

// Run ...
func (t *Task) Run() {
	err := t.f()
	if err != nil {
		metrics.Counter("task-执行任务出错").Incr()
		log.Errorf("Task Run: executing task error. err=%v", err)
	}
}

package taskqueue

import (
	"git.code.oa.com/trpc-go/trpc-go"
	"git.code.oa.com/trpc-go/trpc-go/log"
	"git.code.oa.com/trpc-go/trpc-go/metrics"
)

const (
	// DefaultWorkerNum ...
	DefaultWorkerNum = 10
	// DefaultQueueSize ...
	DefaultQueueSize = 10000
)

// GlobalTaskQueue ...
var GlobalTaskQueue = NewTaskQueue(DefaultWorkerNum, DefaultQueueSize)

// ITaskQueue ...
type ITaskQueue interface {
	// 提交一个任务到队列中
	Submit(ITask)
}

// TaskQueue ...
type TaskQueue struct {
	// 内部的Task队列
	TaskQueue chan ITask
	// 最大的worker数量
	workerNum int
}

// NewTaskQueue ...
func NewTaskQueue(workerNum int, queueSize int) *TaskQueue {
	t := &TaskQueue{
		TaskQueue: make(chan ITask, queueSize),
		workerNum: workerNum,
	}
	t.startWorkers()
	return t
}

// 启动所有工作协程
func (p *TaskQueue) startWorkers() {
	for i := 0; i < p.workerNum; i++ {
		go p.startOneWorker(i)
	}
}

// startOneWorker
func (p *TaskQueue) startOneWorker(workerId int) {
	for task := range p.TaskQueue {
		task.Run()
		log.InfoContextf(trpc.BackgroundContext(),
			"TaskQueue: workerId %v execute task: %v", workerId, task)
	}
}

// Submit ...
func (p *TaskQueue) Submit(task ITask) {
	select {
	case p.TaskQueue <- task:
	default:
		metrics.Counter("task-队列已满丢弃任务").Incr()
		log.ErrorContextf(trpc.BackgroundContext(),
			"TaskQueue Submit: queue is full, discard task %v", task)
	}
}

4. 参考

• Golang 开发需要协程池吗？ - 知乎
• ThinkerQAQ/go_pool
• Strike Freedom