运维开发网

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

运维开发网 https://www.qedev.com 2022-04-01 02:51 出处:网络 作者: u010278923
其实golang用一个函数可以构建一个并发队列,现在编写一个灵活可控的队列程序先定义一个工作typeWorkerstruct{IDintRepJobschanin...

其实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
}

这个里面定义了一个工作队列的切片,可以自定义工作队列的个数,甚至后http://www.cppcns.com期还可以添加work,还定义了一个队列类型的管道。

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

func InitWorkerPool() error {
    n := 3
    WorkerPool = &workerPool{
        workerChan: make(chhttp://www.cppcns.coman *Worker, n),
        workerLiwww.cppcns.comst: 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 {
        

www.cppcns.com

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)

以上为个人经验,希望能给大家一个参考,也希望大家多多支持我们。如有错误或未考虑完全的地方,望不吝赐教。

0

精彩评论

暂无评论...
验证码 换一张
取 消