写在前面
@H_404_2@Go语言作为新兴的语言，最近发展势头很是迅猛，其最大的特点就是原生支持并发。它使用的是“协程（goroutine）模型”，和传统基于 OS 线程和进程实现不同，Go语言的并发是基于用户态的并发，这种并发方式就变得非常轻量，能够轻松运行几万并发逻辑。 @H_404_2@Go 的并发属于 CSP 并发模型的一种实现，CSP 并发模型的核心概念是：“不要通过共享内存来通信，而应该通
过通信来共享内存”。这在 Go 语言中的实现就是 Goroutine 和 Channel。
场景描述
@H_404_2@在一些场景下，有大规模请求（十万或百万级qps），我们处理的请求可能不需要立马知道结果，例如数据的打点，文件的上传等等。这时候我们需要异步化处理。常用的方法有使用resque、MQ、RabbitMQ等。这里我们在Golang语言里进行设计实践。方案演进
- 直接使用goroutine
...
go handle(request)
...
- @H_404_2@goroutine协同带缓存的管道
- 我们定义一个带缓存的管道；
var queue = make(chan job,MAX_QUEUE_SIZE)
- 然后起一个协程处理管道传来的请求；
go func(){
for {
select {
case job := <-queue:
job.Do(request)
case <- quit:
return
}
}
}()
- 接收请求，发送job进行处理
job := &Job{request} queue <- job
@H_404_2@讲真，这种方法使用了缓冲队列一定程度上了提高了并发，但也是治标不治本，大规模并发只是推迟了问题的发生时间。当请求速度远大于队列的处理速度时，缓冲区很快被打满，后面的请求一样被堵塞了。
- job队列+工作池
工作池实现
- 首先，我们定义一个job的接口,具体内容由具体job实现；
type Job interface { Do() error }
- 然后定义一下job队列和work池类型，这里我们work池也用golang的channel实现。
// define job channel
type JobChan chan Job
// define worker channer
type WorkerChan chan JobChan
@H_404_2@我们分别维护一个全局的job队列和工作池。
var ( JobQueue JobChan WorkerPool WorkerChan )
- worker的实现。每一个worker都有一个job channel，在启动worker的时候会被注册到work pool中。启动后通过自身的job channel取到job并执行job。
type Worker struct {
JobChannel JobChan
quit chan bool
}
func (w *Worker) Start() {
go func() {
for {
// regist current job channel to worker pool
WorkerPool <- w.JobChannel
select {
case job := <-w.JobChannel:
if err := job.Do(); err != nil {
fmt.printf("excute job Failed with err: %v",err)
}
// recieve quit event,stop worker
case <-w.quit:
return
}
}
}()
}
- 实现一个分发器（Dispatcher）。分发器包含一个worker的指针数组，启动时实例化并启动最大数目的worker，然后从job队列中不断取job选择可用的worker来执行job。
type Dispatcher struct {
Workers []*Worker
quit chan bool
}
func (d *Dispatcher) Run() {
for i := 0; i < MaxWorkerPoolSize; i++ {
worker := NewWorker()
d.Workers = append(d.Workers,worker)
worker.Start()
}
for {
select {
case job := <-JobQueue:
go func(job Job) {
jobChan := <-WorkerPool
jobChan <- job
}(job)
// stop dispatcher
case <-d.quit:
return
}
}
}
