Golang工作池如何使用

今天小編給大家分享一下Golang工作池如何使用的相關(guān)知識點，內(nèi)容詳細，邏輯清晰，相信大部分人都還太了解這方面的知識，所以分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后有所收獲，下面我們一起來了解一下吧。

一、概念

我們可以將工作池理解為線程池。線程池的創(chuàng)建和銷毀非常消耗資源，所以專門寫一個pool，每次用過的線程池再放回pool中而不是銷毀。不過在Go語言中不會使用系統(tǒng)的線程，而是使用goroutine。gorotine的創(chuàng)建和銷毀比系統(tǒng)線程的消耗要小的多，而且goroutine沒有標號。所以goroutine的pool就不再時線程池，而是work pool（工作池）。

雖然goroutine的系統(tǒng)消耗較小，但也不能隨意在編碼時使用go func()，如果程序頻繁啟動goroutine，會造成極其不可控性能問題。對于可以提前預(yù)知的大量異步處理的任務(wù)就要考慮使用工作池。

工作池的作用控制goroutine的規(guī)模，或者說是goroutine的數(shù)量。在Go語言中，控制goroutine的數(shù)量最好方式就是使用緩存通道。

二、實例

1.簡單示例

下面是Go語言解決工作池的經(jīng)典用法。

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() {
    // 為了使用我們的工作池，我們需要發(fā)送工作和接受工作的結(jié)果，
    // 這里我們定義兩個通道，一個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)
	}
	// 創(chuàng)建5個任務(wù)
	for job := 1; job <= 5; job++ {
		jobs <- job
	}
	close(jobs)
	// 輸出結(jié)果
	for i := 1; i <= 5; i++ {
		<-results
	}
}

上述代碼工作池思想主要體現(xiàn)在jobs的通道上，因為定義了一個緩存長度為100的通道，所以在通道到100以后，新任務(wù)就會阻塞，只有等worker從通道取走一個工作以后才能繼續(xù)分配新工作。

本案例較為簡單，如果worker的數(shù)量較大，業(yè)務(wù)執(zhí)行時間較長的話，我們需要在程序設(shè)計上將jobs和worker的模式進行優(yōu)化，每個worker處理一項工作，工作池可以自定義最大數(shù)量的worker；這樣可以保證goroutine的最大數(shù)量，可程序更加可控，避免代碼消耗壓垮系統(tǒng)。

2.讀入數(shù)據(jù)

下面時改良之后代碼

1package main
import (
	"fmt"
	"reflect"
	"time"
)
// Job 任務(wù)內(nèi)容
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     // 結(jié)束信號
}
var (
	MaxWorker = 5                 // 最大worker數(shù)量
	JobQueue  = make(chan Job, 5) // 工作通道，模擬需處理的工作
)
// Scheduler 排程中心
type Scheduler struct {
	WorkerPool   chan chan Job // 工作池
	WorkerMaxNum int           // 最大工作者數(shù)
	Workers      []*Worker     // worker隊列
}
// NewScheduler 創(chuàng)建排程中心
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 工作池的關(guān)閉
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 監(jiān)聽任務(wù)和結(jié)束信號
func (w Worker) Start() {
	go func() {
		for {
			select {
			case job := <-w.JobChannel: // 收到任務(wù)
				fmt.Println("get a job from private w.JobChannel")
				fmt.Println(job)
			case <-w.exit: // 收到結(jié)束信號
				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任務(wù)
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)
	}
}

定義了兩個結(jié)構(gòu)體：Task任務(wù)和Job工作，Task并沒有實質(zhì)性的內(nèi)容，這里僅僅定義了一個整型變量；

定義兩個全局變量：MaxWorker是最大的worker數(shù)量；JobQueue是Job的通道。這兩個變量都用于后面的模擬，在真實場景中可以不設(shè)置這兩個變量。

定義了一個Worker結(jié)構(gòu)體，與上一個簡單工作池的示例不同，本例的Worker不再是簡單的一個goroutine，而是一個結(jié)構(gòu)體。結(jié)構(gòu)體內(nèi)定義了如下四個變量。?id：worker編號。?exit：這是一個bool類型的通道，當有數(shù)據(jù)寫入時worker結(jié)束運行。?JobChannel：Job類型的通道，該通道是專屬于當前worker的私有工作隊列。?WorkerPool：注意看，定義的時候使用了兩個Channel，每一個元素是一個Job通道，其實每一個元素是一個JobChannel。

NewWorker方法用于創(chuàng)建一個新的worker，要注意該方法的參數(shù)workerPool用于創(chuàng)建worker時傳入，這就說明每個worker與其他worker的WorkerPool是共享的，或者說多個worker使用一個WorkerPool。這一點很重要，這是本示例代碼在上一個簡單示例代碼基礎(chǔ)上的優(yōu)化。而JobChannel和exit變量則是隨著Worker的新建而新建的。

Worker的Start方法，該方法用于監(jiān)聽任務(wù)或者結(jié)束信號。Start方法一開始就用goroutine運行一個匿名函數(shù)，而函數(shù)內(nèi)部是一個無限循環(huán)。在循環(huán)內(nèi)部，首先是把當前的JobChannel注冊到WorkerPool里，一旦注冊進去也就說明該worker可以接收任務(wù)了。然后通過select判斷JobChannel是否可以讀取，也就是其中是否有Job，或者exit通道是否可以讀取。如果JobChannel可讀取，證明有Job，后續(xù)開始處理Job；而如果exit可讀，則結(jié)束當前的無限循環(huán)。所以，后面的代碼中要特別注意對WorkerPool的操作，Worker是從WorkerPool領(lǐng)取工作的。Worker的Stop方法，用于為exit通道寫入數(shù)據(jù)，在Start方法內(nèi)Worker會讀取到寫入的數(shù)據(jù)，進而結(jié)束無限循環(huán)。

NewScheduler函數(shù)用于創(chuàng)建一個Scheduler，可以看到函數(shù)內(nèi)部的WorkerPool是通過make函數(shù)新建的，NewWorker函數(shù)一樣靠參數(shù)傳入。注意WorkerPool是有緩存通道的，緩存長度是MaxWorkers。

Scheduler的Create方法，該方法根據(jù)MaxWorkers最大數(shù)創(chuàng)建Worker，并且把引用存入Workers切片。創(chuàng)建好Worker后，馬上調(diào)用Worker的Start方法，最后通過goroutine運行Schedule方法。Scheduler的Shutdown方法，用于關(guān)閉工作池，調(diào)用所有worker的Stop方法并且關(guān)閉WorkerPool工作池。

Scheduler的Schedule方法，該方法內(nèi)也是一個無限循環(huán)，循環(huán)內(nèi)部就是不停地讀取JobQueue，然后運行一個goroutine。在新運行的goroutine內(nèi)從s.WorkerPool讀取一個JobChannel，注意，Worker注冊到WorkerPool以后此處才可以讀取到，如果WorkerPool的緩存通道內(nèi)沒有JobChannel，則會阻塞，直到讀取到JobChannel，才把Job寫入。

以上就是“Golang工作池如何使用”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家閱讀完這篇文章都有很大的收獲，小編每天都會為大家更新不同的知識，如果還想學(xué)習(xí)更多的知識，請關(guān)注億速云行業(yè)資訊頻道。