分布式后台任务队列模拟(Golang)

第一次,站长亲自招 Gopher 了>>>

最近研究了下gowoker,这东西代码少而精,Golang真是很适合实现这类东西。
我去掉参数配置,JSON,Redis这些东西,用goworker的方式做了个最简单的实现。

实现如下功能:
1. worker向JobServer注册可执行的功能
2. JobServer轮询,有job就执行,没有则继续轮询
3. client向JobServer提出任务请求,并传入参数
4. JobServer依请求丢给worker执行(可并发或串行执行)
5. JobServer继续轮询

我弄的这个代码很少,其中队列用数组代替,同时省掉了很多东西,
但保留了其goroutine与channel最基础的实现。

如果想看goworker的,可以参考下我这个,应当可以更快的弄明白goworker。

演示例子及运结果:

//分布式后台任务队列模拟(一)
//author: Xiong Chuan Liang
//date: 2015年3月24日
package main
import (
	"fmt"
	"runtime"
	//"strconv"
	"time"
	"jobserver"
)
func main() {
	runtime.GOMAXPROCS(runtime.NumCPU())
	fmt.Println("分布式后台任务队列模拟(一)...")
	//Job Server
	js := jobserver.NewJobServer()
	//模拟Worker端注册
	js.RegisterWorkerClass("mail", mailWorker)
	js.RegisterWorkerClass("log", sendLogWorker)
	js.RegisterWorkerClass("exception", paincWorker)
	//模拟客户端发送请求
	go func() {
		time.Sleep(time.Second * 2)
		js.Enqueue("mail", "xcl_168@aliyun.com", "sub", "body")
		js.Enqueue("test_notfound", "aaaaaaaaaaaaaaaaaaa")
		js.Enqueue("log", "x.log", "c.log", "l.log")
		//测试jobserver.PARALLEL/ORDER
		//for j := 0; j < 100; j++ {
		//	js.Enqueue("mail", strconv.Itoa(j))
		//}
		time.Sleep(time.Second)
		js.Enqueue("exception", "try{}exception{}")
		time.Sleep(time.Second * 5)
		js.Enqueue("mail", "xcl_168@aliyun.com2", "sub2", "body2")
	}()
	//启动服务,开始轮询
	// StartServer(轮询间隔,执行方式(并发/顺序))
	js.StartServer(time.Second*3, jobserver.ORDER) //PARALLEL
}
func mailWorker(queue string, args ...interface{}) error {
	fmt.Println("......mail() begin......")
	for _, arg := range args {
		fmt.Println(" args:", arg)
	}
	fmt.Println("......mail() end......")
	return nil
}
func sendLogWorker(queue string, args ...interface{}) error {
	fmt.Println("......sendLog() begin......")
	for _, arg := range args {
		fmt.Println(" args:", arg)
	}
	fmt.Println("......sendLog() end......")
	return nil
}
func paincWorker(queue string, args ...interface{}) error {
	fmt.Println("......painc() begin......")
	panic("\n test exception........................ \n")
	fmt.Println("......painc() end......")
	return nil
}
/*
运行结果:
分布式后台任务队列模拟(一)...
[JobServer] [poll] polling
......mail() begin......
 args: xcl_168@aliyun.com
 args: sub
 args: body
......mail() end......
[JobServer] [poll] test_notfound not found
......sendLog() begin......
 args: x.log
 args: c.log
 args: l.log
......sendLog() end......
[JobServer] [poll] polling
......painc() begin......
[JobServer] [run] Panicking
 test exception........................
[JobServer] [poll] polling
[JobServer] [poll] polling
......mail() begin......
 args: xcl_168@aliyun.com2
 args: sub2
 args: body2
......mail() end......
[JobServer] [poll] polling
[JobServer] [poll] polling
[JobServer] [poll] polling
[JobServer] [poll] quit
*/

上面是顺序执行的运行结果,如果要测试并发,可以将上面代码注释部分打开,JobServer执行方式更改为jobserver.PARALLEL,再执行即可。

具体的实现在下面:

下面两个是基本的一些定义:

package jobserver
type workerFunc func(string, ...interface{}) error
type Workers struct {
	workers map[string]workerFunc
}

package jobserver
type OrdType int
const (
	PARALLEL = 1 << iota
	ORDER
)

关键的JobServer的实现:

//分布式后台任务队列模拟(一)
//author: Xiong Chuan Liang
//date: 2015年3月24日
package jobserver
import (
	"fmt"
	"runtime"
	"sync"
	"time"
)
type JobServer struct {
	Workers
	JobQueue []*WorkerClass
	interval time.Duration
	mt sync.Mutex
	ord OrdType
}
func NewJobServer() *JobServer {
	s := &JobServer{}
	s.workers = make(map[string]workerFunc, 0)
	return s
}
func (s *JobServer) RegisterWorkerClass(className string, f workerFunc) int {
	if _, found := s.workers[className]; found {
		return 1
	}
	s.workers[className] = f
	return 0
}
type WorkerClass struct {
	ClassName string
	Args []interface{}
}
func (s *JobServer) Enqueue(className string, args ...interface{}) bool {
	s.mt.Lock()
	w := &WorkerClass{className, args}
	s.JobQueue = append(s.JobQueue, w)
	s.mt.Unlock()
	return true
}
//poller
func (s *JobServer) poll(quit <-chan bool) <-chan *WorkerClass {
	jobs := make(chan *WorkerClass)
	go func() {
		defer close(jobs)
		for {
			switch {
			case s.JobQueue == nil:
				timeout := time.After(time.Second * 2)
				select {
				case <-quit:
					fmt.Println("[JobServer] [poll] quit")
					return
				case <-timeout:
					fmt.Println("[JobServer] [poll] polling")
				}
			default:
				s.mt.Lock()
				j := s.JobQueue[0]
				if len(s.JobQueue)-1 <= 0 {
					s.JobQueue = nil
				} else {
					s.JobQueue = s.JobQueue[1:len(s.JobQueue)]
				}
				s.mt.Unlock()
				select {
				case jobs <- j:
				case <-quit:
					fmt.Println("[JobServer] [poll] quit")
					return
				}
			}
		}
	}()
	return jobs
}
//worker
func (s *JobServer) work(id int, jobs <-chan *WorkerClass, monitor *sync.WaitGroup) {
	monitor.Add(1)
	f := func() {
		defer monitor.Done()
		for job := range jobs {
			if f, found := s.workers[job.ClassName]; found {
				s.run(f, job)
			} else {
				fmt.Println("[JobServer] [poll] ", job.ClassName, " not found")
			}
		}
	}
	switch s.ord {
	case ORDER:
		f()
	default:
		go f()
	}
}
func (s *JobServer) run(f workerFunc, w *WorkerClass) {
	defer func() {
		if r := recover(); r != nil {
			fmt.Printf("[JobServer] [run] Panicking %s\n", fmt.Sprint(r))
		}
	}()
	f(w.ClassName, w.Args...)
}
func (s *JobServer) StartServer(interval time.Duration, ord OrdType) {
	s.interval = interval
	s.ord = ord
	quit := signals()
	jobs := s.poll(quit)
	var monitor sync.WaitGroup
	switch s.ord {
	case ORDER: //顺序执行
		s.work(0, jobs, &monitor)
	default: //并发执行
		concurrency := runtime.NumCPU()
		for id := 0; id < concurrency; id++ {
			s.work(id, jobs, &monitor)
		}
	}
	monitor.Wait()
}

goworker中要复杂的多,但简单来说最主要的就是实现上面的这些东西,我再另增加了个顺序和并发的选项。

这个例子只能在本机跑,其它东西没有。不过配合Redis,其它客户端或其它语言则可以通过Redis来传递参数及实现队列,

把它真正用起来。

MAIL: xcl_168@aliyun.com

BLOG: http://blog.csdn.net/xcl168

---

本文来自:CSDN博客

感谢作者:xcltapestry

查看原文:分布式后台任务队列模拟(Golang)

7125 次点击 ∙ 2 赞

上一篇:golang中image/draw包用法

下一篇:关于go get安装git golang项目时报错的处理办法

后台任务 轮询 runtime redis

• 编辑
• 预览