寫了一個web接口，想高并發(fā)的請求這個接口，進行壓力測試，所以服務(wù)端就實現(xiàn)了一個線程池。
代碼從網(wǎng)上理解了之后寫的。代碼實例
簡單的介紹：
首先實現(xiàn)一個Job接口，只要有方法實現(xiàn)了Do方法即可
定義個分發(fā)器結(jié)構(gòu)體，主要是WorkPool線程池，用于存儲Worker的JobChannel
init的時候，先初始化一個JobQueue隊列，其他的函數(shù)調(diào)用這個線程池的時候，把任務(wù)放在這個隊列即可。
然后Run的時候，創(chuàng)建多個Worker，起初的時候，woker會把自身的JobChannel先注冊到線程池workerPool中，
然后worker.start就是for{select } 阻塞等待JobChannel中的job任務(wù)。
此時又啟一個go d.Dispatcher() ,將JobQueue中的job任務(wù)放在worker的Jobchannle中。這樣上面的for{select} 就可以拿到任務(wù)去執(zhí)行。
注： maxWorkers 是內(nèi)核CPU數(shù)量，本機4核，就是線程池可以放4個JobChannel，所以，在newWorker的時候，就創(chuàng)建了4個Worker來并發(fā)的處理job任務(wù)。
任務(wù)處理
package workPoolimport "fmt"type Worker struct {WorkerPool chan chan JobJobChannel chan JobQuit chan bool}func NewWorker(workpool chan chan Job) *Worker {return &Worker{WorkerPool: workpool,JobChannel: make(chan Job),Quit: make(chan bool)}}func (w *Worker) Start() {go func() {for{w.WorkerPool <-w.JobChannelselect {case job := <-w.JobChannel:if err := job.Do();err !=nil{fmt.Println("exec some failed ....")}case <-w.Quit:return}}}()}func (w *Worker) Stop() {go func() {w.Quit <-true}()}
實現(xiàn)一個分發(fā)器
package workPoolimport "runtime"var(MaxWorkers = runtime.NumCPU()MaxQueue = 512)type Job interface {Do() error}var JobQueue chan Jobtype Dispatcher struct {MaxWorkers intWorkerPool chan chan JobQuit chan bool}func init() {runtime.GOMAXPROCS(MaxWorkers)JobQueue = make(chan Job,MaxQueue)dispatcher := NewDispatcher(MaxWorkers)dispatcher.Run()}func NewDispatcher(maxWorkers int) *Dispatcher {pool := make(chan chan Job,maxWorkers)return &Dispatcher{MaxWorkers: maxWorkers,WorkerPool: pool,Quit: make(chan bool)}}func (d *Dispatcher) Run() {for i:=0;i<d.MaxWorkers;i++{worker := NewWorker(d.WorkerPool)worker.Start()}go d.Dispatcher()}func (d *Dispatcher) Dispatcher() {for {select {case job := <-JobQueue:jobChannel := <-d.WorkerPooljobChannel <- jobcase <-d.Quit:return}}}
main函數(shù)中可以這樣使用
package mainimport ("context_http/workPool""fmt""net/http")type Msg struct {mobile string}func (m *Msg) Do() error {m.mobile = m.mobile+"_test"fmt.Println(m.mobile)return nil}func getMobile(w http.ResponseWriter,r *http.Request) {defer r.Body.Close()r.ParseForm()mobile := r.PostForm.Get("mobile")var work workPool.Jobm := Msg{mobile: mobile}work = &mworkPool.JobQueue <- workstatus := `{"status":"ok"}`w.Write([]byte(status))}func main() {http.HandleFunc("/test",getMobile)err := http.ListenAndServe(":8081",nil)if err !=nil{fmt.Println("server failure :",err)return}}
聯(lián)系客服
