寫在前面

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

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

場(chǎng)景描述

在一些場(chǎng)景下，有大規(guī)模請(qǐng)求（十萬(wàn)或百萬(wàn)級(jí)qps），我們處理的請(qǐng)求可能不需要立馬知道結(jié)果，例如數(shù)據(jù)的打點(diǎn)，文件的上傳等等。這時(shí)候我們需要異步化處理。常用的方法有使用resque、MQ、RabbitMQ等。這里我們?cè)贕olang語(yǔ)言里進(jìn)行設(shè)計(jì)實(shí)踐。

方案演進(jìn)

1. 直接使用goroutine

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

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

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

   • 我們定義一個(gè)帶緩存的管道；

var queue = make(chan job, MAX_QUEUE_SIZE)
1

• 然后起一個(gè)協(xié)程處理管道傳來(lái)的請(qǐng)求；

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

• 接收請(qǐng)求，發(fā)送job進(jìn)行處理

job := &Job{request}

queue <- job
1
2
3

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

1. job隊(duì)列+工作池

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

工作池實(shí)現(xiàn)

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

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

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

// define job channel
type JobChan chan Job

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

我們分別維護(hù)一個(gè)全局的job隊(duì)列和工作池。

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

• worker的實(shí)現(xiàn)。每一個(gè)worker都有一個(gè)job channel，在啟動(dòng)worker的時(shí)候會(huì)被注冊(cè)到work pool中。啟動(dòng)后通過(guò)自身的job channel取到j(luò)ob并執(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

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