轉自：https://blog.csdn.net/jeanphorn/article/details/79018205

寫在前面

Go語言作為新興的語言，最近發展勢頭很是迅猛，其最大的特點就是原生支援併發。它使用的是“協程（goroutine）模型”，和傳統基於 OS 執行緒和程序實現不同，Go 
語言的併發是基於使用者態的併發，這種併發方式就變得非常輕量，能夠輕鬆執行幾萬併發邏輯。

Go 的併發屬於 CSP 併發模型的一種實現，CSP 併發模型的核心概念是：“不要通過共享記憶體來通訊，而應該通 
過通訊來共享記憶體”。這在 Go 語言中的實現就是 Goroutine 和 Channel。

場景描述

在一些場景下，有大規模請求（十萬或百萬級qps），我們處理的請求可能不需要立馬知道結果，例如資料的打點，檔案的上傳等等。這時候我們需要非同步化處理。常用的方法有使用resque、MQ、RabbitMQ等。這裡我們在Golang語言裡進行設計實踐。

方案演進

1. 直接使用goroutine

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

...
go handle(request)
...

• 1
• 2
• 3
• 4

1. goroutine協同帶快取的管道

   • 我們定義一個帶快取的管道；

var queue = make(chan job, MAX_QUEUE_SIZE)

• 1

• 然後起一個協程處理管道傳來的請求；

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

• 接收請求，傳送job進行處理

job := &Job{request}

queue <- job

• 1
• 2
• 3

講真，這種方法使用了緩衝佇列一定程度上了提高了併發，但也是治標不治本，大規模併發只是推遲了問題的發生時間。當請求速度遠大於佇列的處理速度時，緩衝區很快被打滿，後面的請求一樣被堵塞了。

1. job佇列+工作池

只用緩衝佇列不能解決根本問題，這時候我們可以參考一下執行緒池的概念，定一個工作池（協程池），來限定最大goroutine數目。每次來新的job時，從工作池裡取出一個可用的worker來執行job。這樣一來即保障了goroutine的可控性，也儘可能大的提高了併發處理能力。

工作池實現

• 首先，我們定義一個job的介面, 具體內容由具體job實現；

type Job interface {
    Do() error
}

• 1
• 2
• 3

• 然後定義一下job佇列和work池型別，這裡我們work池也用golang的channel實現。

// 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的實現。每一個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
            }
        }
    }()
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22

• 實現一個分發器（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
        }
    }
}

• 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

感謝