使用golang编写一个并发工作队列

其实golang用一个函数可以构建一个并发队列，现在编写一个灵活可控的队列程序

先定义一个工作

type Worker struct {
   ID      int
   RepJobs chan int64
   SM      *SM
   quit    chan bool
}

包含了workid和执行任务的id，上面的SM只是任务具体内容，这个和具体业务相关，大家自己编写自己的SM业务逻辑

然后定义工作池

type workerPool struct {
   workerChan chan *Worker
   workerList []*Worker
}

这个里面定义了一个工作队列的切片，可以自定义工作队列的个数，甚至后期还可以添加work，还定义了一个队列类型的管道。

定义完成过后就可以初始化工作池了

func InitWorkerPool() error {
   n := 3
   WorkerPool = &workerPool{
       workerChan: make(chan *Worker, n),
       workerList: make([]*Worker, 0, n),
   }
   for i := 0; i < n; i++ {
       worker := NewWorker(i)
       WorkerPool.workerList = append(WorkerPool.workerList, worker)
       worker.Start()
       log.Debugf("worker ％d started", worker.ID)
   }
   return nil
}

这个里面我写死了worker的个数是3，当然这个可以通过读取配置文件或者参数传递的方式；这个里面逐一启动work

worker.Start()，这个是关键

func (w *Worker) Start() {
   go func() {
       for {
           WorkerPool.workerChan <- w
           select {
           case jobID := <-w.RepJobs:
               log.Debugf("worker: ％d, will handle job: ％d", w.ID, jobID)
               w.handleRepJob(jobID)
           case q := <-w.quit:
               if q {
                   log.Debugf("worker: ％d, will stop.", w.ID)
                   return
               }
           }
       }
   }()
}

这个就是go 启动一个协程，先把自己放到workerChan中，然后不断从w.RepJobs管道中获取任务并执行，如果执行完成后又把自己放回到队列中。

所以如果你要有任务需要执行，放到这个管道中即可

func Dispatch() {
   for {
       select {
       case job := <-jobQueue:
           go func(jobID int64) {
               println("Trying to dispatch job: ％d", jobID)
               worker := <-WorkerPool.workerChan
               worker.RepJobs <- jobID
           }(job)
       }
   }
}

从管道中拿出一个worker并把任务id放到worker中去执行。

当然你可以停止worker，甚至可以停止job

func (w *Worker) Stop() {
   go func() {
       w.quit <- true
   }()
}
func (wp *workerPool) StopJobs(jobs []int64) {
   log.Debugf("Works working on jobs: ％v will be stopped", jobs)
   for _, id := range jobs {
       for _, w := range wp.workerList {
           if w.SM.JobID == id {
               log.Debugf("found a worker whose job ID is ％d, will try to stop it", id)
               w.SM.Stop(id)
           }
       }
   }
}

补充一下，int64和字符串转换。

string到int

int,err:=strconv.Atoi(string)

string到int64

int64, err := strconv.ParseInt(string, 10, 64)

int到string

string:=strconv.Itoa(int)

int64到string

string:=strconv.FormatInt(int64,10)

以上为个人经验，希望能给大家一个参考

来源：https://chenxy.blog.csdn.net/article/details/72582200