Go并发处理

写了一个web接口，想高并发的请求这个接口，进行压力测试，所以服务端就实现了一个线程池。

代码从网上理解了之后写的。代码实例

简单的介绍：

　　首先实现一个Job接口，只要有方法实现了Do方法即可

　　定义个分发器结构体，主要是WorkPool线程池，用于存储Worker的JobChannel

　　init的时候，先初始化一个JobQueue队列，其他的函数调用这个线程池的时候，把任务放在这个队列即可。

　　然后Run的时候，创建多个Worker，起初的时候，woker会把自身的JobChannel先注册到线程池workerPool中，

　　然后worker.start就是for{select } 阻塞等待JobChannel中的job任务。

　　此时又启一个go d.Dispatcher() ,将JobQueue中的job任务放在worker的Jobchannle中。这样上面的for{select} 就可以拿到任务去执行。

　　

注： maxWorkers 是内核CPU数量，本机4核，就是线程池可以放4个JobChannel，所以，在newWorker的时候，就创建了4个Worker来并发的处理job任务。

 

任务处理

package workPool

import "fmt"

type Worker struct {
WorkerPool chan chan Job
JobChannel chan Job
Quit 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.JobChannel
select {
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
}()
}

　　

 

实现一个分发器

package workPool

import "runtime"

var(
MaxWorkers = runtime.NumCPU()
MaxQueue = 512
)

type Job interface {
Do() error
}

var JobQueue chan Job

type Dispatcher struct {
MaxWorkers int
WorkerPool chan chan Job
Quit 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.WorkerPool
jobChannel <- job
case <-d.Quit:
return
}
}
}

　　

main函数中可以这样使用

package main

import (
"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.Job
m := Msg{mobile: mobile}
work = &m
workPool.JobQueue <- work
status := `{"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
}
}