這篇文章在medium上很火，作者以實(shí)際案例來分析，講得很好。

我們經(jīng)常聽說使用Go的goroutine和channel很容易實(shí)現(xiàn)高并發(fā)，那是不是全部代碼都放在goroutine中運(yùn)行就可以實(shí)現(xiàn)高并發(fā)程序了呢?很顯然并不是。這篇文章將教大家如何一步一步寫出一個(gè)簡單的， 高并發(fā)的Go程序。

正文

我在幾家不同的公司從事反垃圾郵件，防病毒和反惡意軟件的工作超過15年，現(xiàn)在我知道這些系統(tǒng)最終會(huì)因?yàn)槲覀円刻焯幚泶罅繑?shù)據(jù)而變得越來越復(fù)雜。

目前，我是smsjunk.com的CEO和 KnowBe4的***架構(gòu)師，他們都是網(wǎng)絡(luò)安全行業(yè)的公司。

有趣的是，在過去的10年里，作為一名軟件工程師，我參與過的所有Web后端開發(fā)大部分都是使用RubyonRails完成的。不要誤會(huì)我的意思，我喜歡 RubyonRails，我相信這是一個(gè)了不起的生態(tài)，但是過了一段時(shí)間，你開始以 Ruby的方式思考和設(shè)計(jì)系統(tǒng)，忘了如何高效和原本可以利用多線程、并行、快速執(zhí)行和小的內(nèi)存消耗來簡化軟件架構(gòu)。多年來，我是一名C/C++，Delphi和 C#開發(fā)人員，而且我剛開始意識(shí)到如何正確的使用工具進(jìn)行工作可能會(huì)有多復(fù)雜。

我對(duì)互聯(lián)網(wǎng)中那些語言和框架戰(zhàn)爭(zhēng)并不太感興趣，比如哪門語言更好，哪個(gè)框架更快。 我始終相信效率，生產(chǎn)力和代碼可維護(hù)性主要取決于如何簡單的構(gòu)建解決方案。

問題

在處理我們的匿名監(jiān)測(cè)和分析系統(tǒng)時(shí)，我們的目標(biāo)是能夠處理來自數(shù)百萬端點(diǎn)的大量POST請(qǐng)求。Web處理程序?qū)⑹盏揭粋€(gè)JSON文檔，該文檔可能包含需要寫入 AmazonS3的多個(gè)有效內(nèi)容的集合，以便我們的 map-reduce系統(tǒng)稍后對(duì)這些數(shù)據(jù)進(jìn)行操作。

傳統(tǒng)上，我們會(huì)考慮創(chuàng)建一個(gè)工作層架構(gòu)，利用諸如以下的技術(shù)棧：

• Sidekiq
• Resque
• DelayedJob
• ElasticbeanstalkWorkerTier
• RabbitMQ
• ...

并搭建2個(gè)不同的集群，一個(gè)用于web前端，一個(gè)用于worker，因此我們可以隨意擴(kuò)容機(jī)器來處理即將到來的請(qǐng)求。

從一開始，我們的團(tuán)隊(duì)就知道我們可以在Go中這樣做，因?yàn)樵谟懻撾A段我們看到這可能是一個(gè)非常大流量的系統(tǒng)。我一直在使用Go，大約快2年時(shí)間了，而且我們也使用Go開發(fā)了一些系統(tǒng)，但是沒有一個(gè)系統(tǒng)的流量能夠達(dá)到這個(gè)數(shù)量級(jí)。我們首先創(chuàng)建了幾個(gè)struct來定義我們通過POST調(diào)用接收到的Web請(qǐng)求，并將其上傳到S3存儲(chǔ)中。

type PayloadCollection struct { 
    WindowsVersion  string    `json:"version"` 
    Token           string    `json:"token"` 
    Payloads        []Payload `json:"data"` 
} 
 
type Payload struct { 
    // [redacted] 
} 
 
func (p *Payload) UploadToS3() error { 
    // the storageFolder method ensures that there are no name collision in 
    // case we get same timestamp in the key name 
    storage_path := fmt.Sprintf("%v/%v", p.storageFolder, time.Now().UnixNano()) 
 
    bucket := S3Bucket 
 
    b := new(bytes.Buffer) 
    encodeErr := json.NewEncoder(b).Encode(payload) 
    if encodeErr != nil { 
        return encodeErr 
    } 
 
    // Everything we post to the S3 bucket should be marked 'private' 
    var acl = s3.Private 
    var contentType = "application/octet-stream" 
 
    return bucket.PutReader(storage_path, b, int64(b.Len()), contentType, acl, s3.Options{}) 
} 

Naive的做法-硬核使用Goroutine

最初，我們對(duì)POST處理程序進(jìn)行了非常簡單粗暴的實(shí)現(xiàn)，將每個(gè)請(qǐng)求直接放到新的goroutine中運(yùn)行：

func payloadHandler(w http.ResponseWriter, r *http.Request) { 
 
    if r.Method != "POST" { 
        w.WriteHeader(http.StatusMethodNotAllowed) 
        return 
    } 
 
    // Read the body into a string for json decoding 
    var content = &PayloadCollection{} 
    err := json.NewDecoder(io.LimitReader(r.Body, MaxLength)).Decode(&content) 
    if err != nil { 
        w.Header().Set("Content-Type", "application/json; charset=UTF-8") 
        w.WriteHeader(http.StatusBadRequest) 
        return 
    } 
     
    // Go through each payload and queue items individually to be posted to S3 
    for _, payload := range content.Payloads { 
        go payload.UploadToS3()   // <----- DON'T DO THIS 
    } 
 
    w.WriteHeader(http.StatusOK) 
} 

對(duì)于一般的并發(fā)量，這其實(shí)是可行的，但這很快就證明不能適用于高并發(fā)場(chǎng)景。我們可能有更多的請(qǐng)求，當(dāng)我們將***個(gè)版本部署到生產(chǎn)環(huán)境時(shí)，我們開始看到的數(shù)量級(jí)并不是如此，我們低估了并發(fā)量。

上述的方法有幾個(gè)問題。沒有辦法控制正在工作的goroutine的數(shù)量。而且，由于我們每分鐘有100萬個(gè)POST請(qǐng)求，所以系統(tǒng)很快就崩潰了。

重來

我們需要找到另一種的方法。從一開始我們就開始討論如何讓請(qǐng)求處理程序的生命周期盡可能的短，并在后臺(tái)產(chǎn)生處理。當(dāng)然，這是在 RubyonRails必須要做的事情，否則，不管你是使用puma，unicorn還是 passenger，你的所有的可用的web worker都將阻塞。

那么我們就需要利用常見的解決方案來完成這項(xiàng)工作，比如Resque，Sidekiq， SQS等。當(dāng)然還有其他工具，因?yàn)橛泻芏喾椒梢詫?shí)現(xiàn)。

因此，我們第二次改進(jìn)是創(chuàng)建一個(gè)buffer channel，我們可以將一些作業(yè)請(qǐng)求扔進(jìn)隊(duì)列并將它們上傳到S3，由于我們可以控制隊(duì)列的***長度，并且有足夠的RAM來排隊(duì)處理內(nèi)存中的作業(yè)，因此我們認(rèn)為只要在通道隊(duì)列中緩沖作業(yè)就行了。

var Queue chan Payload 
 
func init() { 
    Queue = make(chan Payload, MAX_QUEUE) 
} 
 
func payloadHandler(w http.ResponseWriter, r *http.Request) { 
    ... 
    // Go through each payload and queue items individually to be posted to S3 
    for _, payload := range content.Payloads { 
        Queue <- payload 
    } 
    ... 
} 

然后，為了將任務(wù)從buffer channel中取出并處理它們，我們正在使用這樣的方式:

func StartProcessor() { 
    for { 
        select { 
        case job := <-Queue: 
            job.payload.UploadToS3()  // <-- STILL NOT GOOD 
        } 
    } 
} 

說實(shí)話，我不知道我們?cè)谙胧裁?，這肯定是一個(gè)難熬的夜晚。這種方法并沒有給我們帶來什么提升，我們用一個(gè)緩沖的隊(duì)列替換了有缺陷的并發(fā)，也只是推遲了問題的產(chǎn)生時(shí)間而已。我們的同步處理器每次只向S3上傳一個(gè)有效載荷，由于傳入請(qǐng)求的速率遠(yuǎn)遠(yuǎn)大于單個(gè)處理器上傳到S3的能力，因此我們的buffer channel迅速達(dá)到極限，隊(duì)列已經(jīng)阻塞并且無法再往里邊添加作業(yè)。

我們只是簡單的繞過了這個(gè)問題，最終導(dǎo)致我們的系統(tǒng)完全崩潰。在我們部署這個(gè)有缺陷的版本后，我們的延遲持續(xù)的升高。

更好的解決方案

我們決定在Go channel上使用一個(gè)通用模式來創(chuàng)建一個(gè) 2-tier(雙重)channel系統(tǒng)，一個(gè)用來處理排隊(duì)的job，一個(gè)用來控制有多少worker在 JobQueue上并發(fā)工作。

這個(gè)想法是將上傳到S3的并行速度提高到一個(gè)可持續(xù)的速度，同時(shí)不會(huì)造成機(jī)器癱瘓，也不會(huì)引發(fā)S3的連接錯(cuò)誤。

所以我們選擇創(chuàng)建一個(gè) Job/Worker模式。對(duì)于那些熟悉Java，C#等的人來說，可以將其視為Golang使用channel來實(shí)現(xiàn)WorkerThread-Pool的方式。

var ( 
    MaxWorker = os.Getenv("MAX_WORKERS") 
    MaxQueue  = os.Getenv("MAX_QUEUE") 
) 
 
// Job represents the job to be run 
type Job struct { 
    Payload Payload 
} 
 
// A buffered channel that we can send work requests on. 
var JobQueue chan Job 
 
// Worker represents the worker that executes the job 
type Worker struct { 
    WorkerPool  chan chan Job 
    JobChannel  chan Job 
    quit        chan bool 
} 
 
func NewWorker(workerPool chan chan Job) Worker { 
    return Worker{ 
        WorkerPool: workerPool, 
        JobChannel: make(chan Job), 
        quit:       make(chan bool)} 
} 
 
// Start method starts the run loop for the worker, listening for a quit channel in 
// case we need to stop it 
func (w Worker) Start() { 
    go func() { 
        for { 
            // register the current worker into the worker queue. 
            w.WorkerPool <- w.JobChannel 
 
            select { 
            case job := <-w.JobChannel: 
                // we have received a work request. 
                if err := job.Payload.UploadToS3(); err != nil { 
                    log.Errorf("Error uploading to S3: %s", err.Error()) 
                } 
 
            case <-w.quit: 
                // we have received a signal to stop 
                return 
            } 
        } 
    }() 
} 
 
// Stop signals the worker to stop listening for work requests. 
func (w Worker) Stop() { 
    go func() { 
        w.quit <- true 
    }() 
} 

我們修改了我們的Web請(qǐng)求處理程序以創(chuàng)建具有有效負(fù)載的Job struct，并將其發(fā)送到 JobQueueChannel以供worker處理。

func payloadHandler(w http.ResponseWriter, r *http.Request) { 
 
    if r.Method != "POST" { 
        w.WriteHeader(http.StatusMethodNotAllowed) 
        return 
    } 
 
    // Read the body into a string for json decoding 
    var content = &PayloadCollection{} 
    err := json.NewDecoder(io.LimitReader(r.Body, MaxLength)).Decode(&content) 
    if err != nil { 
        w.Header().Set("Content-Type", "application/json; charset=UTF-8") 
        w.WriteHeader(http.StatusBadRequest) 
        return 
    } 
 
    // Go through each payload and queue items individually to be posted to S3 
    for _, payload := range content.Payloads { 
 
        // let's create a job with the payload 
        work := Job{Payload: payload} 
 
        // Push the work onto the queue. 
        JobQueue <- work 
    } 
 
    w.WriteHeader(http.StatusOK) 
} 

在我們的Web服務(wù)器初始化期間，我們創(chuàng)建一個(gè)Dispatcher并調(diào)用Run()來創(chuàng)建worker池并開始監(jiān)聽JobQueue中出現(xiàn)的Job。

dispatcher := NewDispatcher(MaxWorker)  
dispatcher.Run() 

以下是我們調(diào)度程序?qū)崿F(xiàn)的代碼：

type Dispatcher struct { 
    // A pool of workers channels that are registered with the dispatcher 
    WorkerPool chan chan Job 
} 
 
func NewDispatcher(maxWorkers int) *Dispatcher { 
    pool := make(chan chan Job, maxWorkers) 
    return &Dispatcher{WorkerPool: pool} 
} 
 
func (d *Dispatcher) Run() { 
    // starting n number of workers 
    for i := 0; i < d.maxWorkers; i++ { 
        worker := NewWorker(d.pool) 
        worker.Start() 
    } 
 
    go d.dispatch() 
} 
 
func (d *Dispatcher) dispatch() { 
    for { 
        select { 
        case job := <-JobQueue: 
            // a job request has been received 
            go func(job Job) { 
                // try to obtain a worker job channel that is available. 
                // this will block until a worker is idle 
                jobChannel := <-d.WorkerPool 
 
                // dispatch the job to the worker job channel 
                jobChannel <- job 
            }(job) 
        } 
    } 
} 

請(qǐng)注意，我們實(shí)例化了***數(shù)量的worker，并將其保存到worker池中(就是上面的 WorkerPoolChannel)。由于我們已經(jīng)將Amazon Elasticbeanstalk用于Docker化的Go項(xiàng)目，并且我們始終嘗試遵循12要素方法來配置生產(chǎn)中的系統(tǒng)，因此我們從環(huán)境變量中讀取這些值，這樣我們就可以快速調(diào)整這些值以控制工作隊(duì)列的數(shù)量和***規(guī)模，而不需要重新部署集群。

var (  
  MaxWorker = os.Getenv("MAX_WORKERS")  
  MaxQueue  = os.Getenv("MAX_QUEUE")  
) 

在我們發(fā)布了這個(gè)版本之后，我們立即看到我們的所有的請(qǐng)求延遲都下降到了一個(gè)很低的數(shù)字，我們處理請(qǐng)求的效率大大提升。

在我們的彈性負(fù)載均衡器完全熱身之后的幾分鐘，我們看到我們的ElasticBeanstalk應(yīng)用程序每分鐘提供近100萬次請(qǐng)求。通常在早晨的幾個(gè)小時(shí)里，流量高峰會(huì)超過每分鐘100萬個(gè)請(qǐng)求。

我們部署了新的代碼，服務(wù)器的數(shù)量從100臺(tái)減少到大約20臺(tái)。

在恰當(dāng)?shù)嘏渲昧思汉妥詣?dòng)縮放設(shè)置以后，我們?cè)谏森h(huán)境用4臺(tái)EC2 c4就能完成工作了。如果CPU在連續(xù)5分鐘內(nèi)超過90%，彈性自動(dòng)縮放系統(tǒng)就自動(dòng)擴(kuò)容一個(gè)新的實(shí)例。

結(jié)論

簡單總是我的制勝法寶。我們可以設(shè)計(jì)一個(gè)擁有多隊(duì)列，多后臺(tái)進(jìn)程和難以部署的復(fù)雜系統(tǒng)，但是相反我們決定利用Elasticbeanstalk的自動(dòng)縮放和高效簡單的方式去并發(fā)，Go語言很好的提供了這些功能。

經(jīng)驗(yàn)告訴我們，用最合適的工具去完成工作。有時(shí)，當(dāng)你的 RubyonRails系統(tǒng)需要實(shí)現(xiàn)一個(gè)非常強(qiáng)大的處理程序時(shí)，可以考慮在 Ruby生態(tài)系統(tǒng)之外尋找更簡單且更強(qiáng)大的替代解決方案。