Golang工作池的使用实例讲解

func worker(id int, jobs <-chan int, results chan<- int) {
	for job := range jobs {
		fmt.Printf("worker(%d) start to do job(%d)\n", id, job)
		time.Sleep(time.Second)
		fmt.Printf("worker(%d) finished job(%d)\n", id, job)
		results <- job
	}
}
func main() {
    // 为了使用我们的工作池，我们需要发送工作和接受工作的结果，
    // 这里我们定义两个通道，一个jobs，一个results
	jobs := make(chan int, 100)
	results := make(chan int, 100)
	// 开启3个goroutine
	for id := 1; id <= 3; id++ {
		go worker(id, jobs, results)
	}
	// 创建5个任务
	for job := 1; job <= 5; job++ {
		jobs <- job
	}
	close(jobs)
	// 输出结果
	for i := 1; i <= 5; i++ {
		<-results
	}
}

1package main
import (
	"fmt"
	"reflect"
	"time"
)
// Job 任务内容
type Job struct {
	ID   int
	Name string
}
// Worker 工作
type Worker struct {
	id         int           // id
	WorkerPool chan chan Job // 工作者池（通道的通道），每个元素都是一个job通道， 公共的job
	JobChannel chan Job      // 工作通道，每个元素是一个job，worker私有的job
	exit       chan bool     // 结束信号
}
var (
	MaxWorker = 5                 // 最大worker数量
	JobQueue  = make(chan Job, 5) // 工作通道，模拟需处理的工作
)
// Scheduler 排程中心
type Scheduler struct {
	WorkerPool   chan chan Job // 工作池
	WorkerMaxNum int           // 最大工作者数
	Workers      []*Worker     // worker队列
}
// NewScheduler 创建排程中心
func NewScheduler(workerMaxNum int) *Scheduler {
	workerPool := make(chan chan Job, workerMaxNum) // 工作池
	return &Scheduler{WorkerPool: workerPool, WorkerMaxNum: workerMaxNum}
}
// Start 工作池开始
func (s *Scheduler) Start() {
	Workers := make([]*Worker, s.WorkerMaxNum)
	for i := 0; i < s.WorkerMaxNum; i++ {
		worker := NewWorker(s.WorkerPool, i)
		worker.Start()
		Workers[i] = &worker
	}
	s.Workers = Workers
	go s.schedule()
}
// Stop 工作池的关闭
func (s *Scheduler) Stop() {
	Workers := s.Workers
	for _, w := range Workers {
		w.Stop()
	}
	time.Sleep(time.Second)
	close(s.WorkerPool)
}
func NewWorker(WorkerPool chan chan Job, id int) Worker {
	fmt.Printf("new a worker(%d)\n", id)
	return Worker{
		id:         id,
		WorkerPool: WorkerPool,
		JobChannel: make(chan Job),
		exit:       make(chan bool),
	}
}
// Start 监听任务和结束信号
func (w Worker) Start() {
	go func() {
		for {
			select {
			case job := <-w.JobChannel: // 收到任务
				fmt.Println("get a job from private w.JobChannel")
				fmt.Println(job)
			case <-w.exit: // 收到结束信号
				fmt.Println("worker exit", w)
				return
			}
		}
	}()
}
func (w Worker) Stop() {
	go func() {
		w.exit <- true
	}()
}
// 排程
func (s *Scheduler) schedule() {
	for {
		select {
		case job := <-JobQueue:
			fmt.Println("get a job from JobQueue")
			go func(job Job) {
				//从WorkerPool获取jobChannel，忙时阻塞
				jobChannel := <-s.WorkerPool
				fmt.Println("get a private jobChannel from public s.WorkerPool", reflect.TypeOf(jobChannel))
				jobChannel <- job
				fmt.Println("worker's private jobChannel add one job")
			}(job)
		}
	}
}
func main() {
	scheduler := NewScheduler(MaxWorker)
	scheduler.Start()
	jobQueue()
	scheduler.Stop()
}
// 模拟Job任务
func jobQueue() {
	for i := 1; i <= 30; i++ {
		JobQueue <- Job{ID: i, Name: fmt.Sprintf("Job【%d】", i)}
		fmt.Printf("jobQueue add %d job\n", i)
	}
}