寫在前面

Go語言作為新興的語言，最近發(fā)展勢頭很是迅猛，其最大的特點就是原生支持并發(fā)。它使用的是“協(xié)程（goroutine）模型”，和傳統(tǒng)基于 OS 線程和進程實現(xiàn)不同，Go
語言的并發(fā)是基于用戶態(tài)的并發(fā)，這種并發(fā)方式就變得非常輕量，能夠輕松運行幾萬并發(fā)邏輯。

Go 的并發(fā)屬于 CSP 并發(fā)模型的一種實現(xiàn)，CSP 并發(fā)模型的核心概念是：“不要通過共享內存來通信，而應該通
過通信來共享內存”。這在 Go 語言中的實現(xiàn)就是 Goroutine 和 Channel。

場景描述

在一些場景下，有大規(guī)模請求（十萬或百萬級qps），我們處理的請求可能不需要立馬知道結果，例如數(shù)據(jù)的打點，文件的上傳等等。這時候我們需要異步化處理。常用的方法有使用resque、MQ、RabbitMQ等。這里我們在Golang語言里進行設計實踐。

方案演進

1. 直接使用goroutine

在Go語言原生并發(fā)的支持下，我們可以直接使用一個goroutine（如下方式）去并行處理這個請求。但是，這種方法明顯有些不好的地方，我們沒法控制goroutine產生數(shù)量，如果處理程序稍微耗時，在單機萬級十萬級qps請求下，goroutine大規(guī)模爆發(fā)，內存暴漲，處理效率會很快下降甚至引發(fā)程序崩潰。

...
go handle(request)
...
1
2
3
4

1. goroutine協(xié)同帶緩存的管道

   • 我們定義一個帶緩存的管道；

var queue = make(chan job, MAX_QUEUE_SIZE)
1

• 然后起一個協(xié)程處理管道傳來的請求；

go func(){
   for {
    select {
        case job := <-queue:
            job.Do(request)
        case <- quit:
            return
    }

   }
}()
1
2
3
4
5
6
7
8
9
10
11
12

• 接收請求，發(fā)送job進行處理

job := &Job{request}

queue <- job
1
2
3

講真，這種方法使用了緩沖隊列一定程度上了提高了并發(fā)，但也是治標不治本，大規(guī)模并發(fā)只是推遲了問題的發(fā)生時間。當請求速度遠大于隊列的處理速度時，緩沖區(qū)很快被打滿，后面的請求一樣被堵塞了。

1. job隊列+工作池

只用緩沖隊列不能解決根本問題，這時候我們可以參考一下線程池的概念，定一個工作池（協(xié)程池），來限定最大goroutine數(shù)目。每次來新的job時，從工作池里取出一個可用的worker來執(zhí)行job。這樣一來即保障了goroutine的可控性，也盡可能大的提高了并發(fā)處理能力。

工作池實現(xiàn)

• 首先，我們定義一個job的接口, 具體內容由具體job實現(xiàn)；

type Job interface {
    Do() error
}
1
2
3

• 然后定義一下job隊列和work池類型，這里我們work池也用golang的channel實現(xiàn)。

// define job channel
type JobChan chan Job

// define worker channer
type WorkerChan chan JobChan
1
2
3
4
5
6

我們分別維護一個全局的job隊列和工作池。

var (
    JobQueue          JobChan
    WorkerPool        WorkerChan
)
1
2
3
4
5

• worker的實現(xiàn)。每一個worker都有一個job channel，在啟動worker的時候會被注冊到work pool中。啟動后通過自身的job channel取到job并執(zhí)行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
            }
        }
    }()
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

• 實現(xiàn)一個分發(fā)器（Dispatcher）。分發(fā)器包含一個worker的指針數(shù)組，啟動時實例化并啟動最大數(shù)目的worker，然后從job隊列中不斷取job選擇可用的worker來執(zhí)行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
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

感謝

感謝Handling 1 Million Requests per Minute with Go這篇文章給予的巨大啟發(fā)。