用Golang處理每分鐘百萬級請求

來自專欄技術譯文

翻譯原文鏈接 轉帖／轉載請註明出處

原文鏈接@medium.com 發表於2017/08/30

我在防垃圾郵件，防病毒和防惡意軟體領域已經工作了15年，前後在好幾個公司任職。我知道這些系統最後都會因為要處理海量的數據而變得非常複雜。

我現在是smsjunk.com的CEO並且是KnowBe4的首席架構師。這兩個公司在網路安全領域都非常活躍。

有趣的是，在過去10年里作為一個碼農，所有我經歷過的網站後台開發用的幾乎都是用Ruby on Rails。不要誤解，我很喜歡`Ruby on Rails`並且認為它是一個非常棒的開發環境。往往在一段時間後，你開始以ruby的方式來設計系統。這時你會忘記利用多線程，並行，快速執行（fast executions）和較小的內存開銷（small memory overhead），軟體的架構會變得簡單而高效。很多年來，我一直是C/C++，Delphi和C#的開發者。我開始意識到使用正確的工具，工作會變得簡單很多。

我對語言和框架並不是很熱衷。我相信效率，產出和代碼的可維護性取決於你如何架構一個簡潔地解決方案。

問題

在開發我們的匿名遙測和分析系統時，我們的目標是能夠處理從上百萬個端點發來的大量POST請求。HTTP請求處理函數會收到包含很多載荷（payloads）的JSON文檔。這些載荷（payloads）需要被寫到Amazon S3上，接著由map-reduce系統來處理。

通常我們會創建一個worker池架構（worker-tier architecture)。利用如下的一些工具：

• Sidekiq
• Resque
• DelayedJob
• Elasticbeanstalk Worker Tier
• RabbitMQ

然後設置兩個集群，一個用作處理HTTP請求，另外一個用作workers。這樣我們能夠根據處理的後台工作量進行擴容。

從一開始我們小組就覺得應該用Go來實現，因為在討論階段我們估計這可能會是一個處理非常大流量的系統。我已經使用Go語言兩年並用它在工作中開發了一些系統，但它們都沒有處理過這麼大的負載（load）。

我們首先創建了幾個結構來定義HTTP請求的載荷。我們通過POST請求接收這些載荷，然後用一個函數上傳到S3 bucket。

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{})}

簡單地使用Goroutines

一開始我們用了最簡單的方法來實現POST請求的處理函數。我們嘗試通過goroutine來並行處理請求。

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() // <----- DONT DO THIS } w.WriteHeader(http.StatusOK)}

對於適量的負載，這個方案應該沒有問題。但是負載增加以後這個方法就不能很好地工作。當我們把這個版本部署到生產環境中後，我們遇到了比預期大一個數量級的請求量。我們完全低估了流量。

這個方法有些不盡如人意。它無法控制創建goroutine的數量。因為我們每分鐘收到了一百萬個POST請求，上面的代碼很快就奔潰了。

再次嘗試

我們需要一個不同的解決方案。在一開始，我們就討論到需要把HTTP請求處理函數寫的簡潔，然後把處理工作轉移到後台。當然，這是你在Ruby on Rails世界裡必須做的，否則你會阻塞所有worker的工作（例如puma，unicorn，passenger等等，我們這裡就不繼續討論JRuby了）。我們需要用到Resque，Sidekiq，SQS等常用的解決方案。這個列表可以很長，因為有許多方法來完成這項任務。

第二個版本是創建帶緩衝的channel。這樣我們可以把工作任務放到隊列里然後再上傳到S3。因為可以控制隊列的長度並且有充足的內存，我們覺得把工作任務緩存在channel隊列里應該沒有問題。

var Queue chan Payloadfunc 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 } ...}

然後我們需要從隊列里提取工作任務並進行處理。代碼下圖所示：

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

坦白的說，我不知道我們當時在想什麼。這肯定是熬夜喝紅牛的結果。這個方法並沒有給我們帶來任何幫助。隊列僅僅是將問題延後了。我們的處理函數（processor）一次僅上傳一個載荷（payload），而接收請求的速率比一個處理函數上傳S3的能力大太多了，帶緩衝的channel很快就到達了它的極限。從而阻塞了HTTP請求處理函數往隊列里添加更多的工作任務。

我們僅僅是延緩了問題的觸發。系統在倒計時，最後還是崩潰了。在這個有問題的版本被部署之後，系統的延遲以恆定速度在不停地增長。

更好的解決辦法

我們決定使用Go channel的常用編程模式。使用一個兩級channel系統，一個用來存放任務隊列，另一個用來控制處理任務隊列的並發量。

這裡的想法是根據一個可持續的速率將S3上傳並行化。這個速率不會使機器變慢或者導致S3的連接錯誤。我們選擇了一個Job／Worker模式。如果你們對Java，C#等語言熟悉的話，可以把它想像成是Go語言用channel來實現的工作線程池。

var ( MaxWorker = os.Getenv("MAX_WORKERS") MaxQueue = os.Getenv("MAX_QUEUE"))// Job represents the job to be runtype 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 jobtype 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 itfunc (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 }()}

我們修改了HTTP請求處理函數來創建一個含有載荷（payload）的Job結構，然後將它送到一個叫JobQueue的channel。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 { // lets create a job with the payload work := Job{Payload: payload} // Push the work onto the queue. JobQueue <- work } w.WriteHeader(http.StatusOK)}

在初始化服務的時候，我們創建了一個Dispatcher並且調用了Run()函數來創建worker池。這些worker會監聽JobQueue上是否有新的任務並進行處理。

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

下面是我們的dispatcher實現代碼：

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) } }}

這裡我們提供了創建worker的最大數目作為參數，並把這些worker加入到worker池裡。因為我們已經在docker化的Go環境里使用了Amazon的Elasticbeanstalk並且嚴格按照12-factor方法來配置我們的生產環境，這些參數值可以從環境變數里獲得。我們可以方便地控制worker數目和任務隊列的長度。我們可以快速地調整這些值而不需要重新部署整個集群。

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

部署了新版本之後，我們看到系統延遲一下子就降到了可以忽略的量級。同時處理請求的能力也大幅攀升。

在Elastic Load Balancers熱身後幾分鐘，我們看到Elasticbeanstalk應用開始處理將近每分鐘一百萬個請求。我們的流量通常在早上的時候會攀升至超過每分鐘一百萬個請求。同時，我們也將伺服器的數目從100台縮減到了20台。

通過合理地配置集群和auto-scaling，我們能夠做到只配置4台EC2 c4.Large實例。然後當CPU使用率持續5分鐘在90%以上時用Elastic Auto-Scaling來創建新的實例。

結束語

對我來說簡潔（simplicity）是第一位的。我們可以利用無數隊列，很多後台worker以及複雜的部署來設計一個複雜系統，最終我們還是使用了Elasticbeanstalk auto-scaling的強大功能和Go語言提供的應對並發的簡單方法。用僅僅4台機器（可能還不如我的MacBook Pro強大）來處理每分鐘一百萬次POST請求對Amazon S3進行寫操作。

每項任務都有對應的正確工具。當你的Ruby on Rails系統需要一個很強大的HTTP請求處理器，可以嘗試看看ruby生態系統以外的其它更強大的選項。