[譯]使用golang每分鐘處理百萬請求
- 2019 年 10 月 3 日
- 筆記
[譯]使用golang每分鐘處理百萬請求
在Malwarebytes，我們正在經歷驚人的增長，自從我在1年前加入矽谷的這家公司以來，我的主要職責是為多個系統做架構和開發，為這家安全公司的快速發展以及百萬日活產品所必需的基礎設施提供支援。我曾在一些不同的公司從事反病毒和反惡意軟體行業超過12年，我知道這些系統最終會因為我們每天處理的大量數據而變得十分複雜。
有趣的是，在過去9年左右的時間裡，我所參與的所有Web後端的開發工作大部分都是在Rails框架的基礎上用ruby實現的。不要誤解我的意思，雖然我很喜歡Rails框架和Ruby，我相信這是一個會讓你感到驚嘆的環境，但一段時間之後，你就會用ruby的方式來進行思考和設計系統，卻忘記了本來可以利用多執行緒，並行化，快速執行和小記憶體開銷來使你的軟體架構變得如此高效和簡單。作為一個多年的C / C ++，Delphi和C＃開發人員，我也同樣開始意識到了如何在工作中使用正確的工具來降低事情的複雜度。
作為首席架構師，我不太重視對互聯網所進行的語言和框架之爭。我相信軟體的效率(efficiency)，生產力(productivity)和程式碼可維護性主要取決於你構建解決方案的簡單程度。
問題
在構建我們的匿名檢測和分析系統時，我們的目標是能夠處理來自數百萬個端點的的大量POST請求。 Web處理程式將接收一個JSON文檔，該文檔可能包含了需要寫入Amazon S3(註：這個是亞馬遜的雲計算服務平台)的許多負載（payload）的集合，以便我們的map-reduce系統稍後對這些數據進行處理。
從傳統上來說，我們會考慮利用以下工具（基本都是開源的）創建一個工作層架構：
- Sidekiq
- Resque
- DelayedJob
- Elasticbeanstalk Worker Tier
- RabbitMQ
- and so on…
然後創建2個不同的集群，一個用於Web前端，另一個用於後台工作的處理，以擴展可以處理的後台工作的數量。
但是從一開始，我們的團隊就知道我們應該使用go語言進行開發，因為在討論階段我們就意識到了這可能是一個吞吐量非常大的系統。我使用go大概有2年左右的時間，我們也開發了一些系統，但是沒有一個系統有如此大的吞吐量。
我們開始創建了一些結構體，用於定義通過POST調用獲取的網路請求負載（payload），同時定義了一個方法將這些負載上傳到S3。
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{}) }從原生方法到Go協程
一開始我們用非常原生的方法來實現POST句柄，嘗試通過使用一個簡單的協程來將作業的處理並行化：
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) }對於適當的載荷，上面的方法在大多數情況下能夠工作，但是在大規模載荷的情況下，上面的方法很快被證明不能夠發揮很好的作用。我們想像會有很多的請求，但是當部署第一個版本到生產環境時，沒有想到會有如此的量級。我們完全低估了載荷的數量。
上述方法在幾個方面都很糟糕。它沒有辦法控制產生的go協程數量。既然我們每分鐘會收到百萬請求，這段程式碼會很快崩潰。
再試一次
我們需要尋找別的方法。開始我們就討論了我們需要保持請求句柄的生命周期的短暫性以及請求的處理要在後台進行。當然，這是在Rails的世界中用Ruby必須要做到的，否則會阻塞所有能用的web工作處理器，不論你是使用puma,unicorn或者passenger。然後我們需要利用常見的解決方案來做到這一點，例如Resque，Sidekiq，SQS等。當然還有很多其它方法也能做到這一點，
所以第二次迭代中我們會創建一個緩衝通道（buffered channel），作業可以插入到緩衝通道中並將作業的負載上傳到S3，因為我們可以控制緩衝通道中元素的數量，並且我們有大量的記憶體將作業插入緩衝通道，我們認為這種方法是沒有任何問題的。
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 } ... }然後從緩衝通道中取出作業進行處理，像下面這樣：
func StartProcessor() { for { select { case job := <-Queue: job.payload.UploadToS3() // <-- STILL NOT GOOD } } }說實話，我不知道我們在想什麼。這個夜晚註定是用紅牛度過的。這種方法並沒有給我們帶來任何好處，我們用緩衝隊列代替了有缺陷的並發，但這只是推遲了問題。我們的同步處理器一次只向S3上傳一個有效載荷（payload），由於傳入請求的速率遠遠大於單個處理器上傳到S3的能力，我們的緩衝通道很快就達到了極限並阻止了請求句柄可以添加更多作業的能力。
我們只是避免了這個問題，系統的死期最終也進入了倒計時。在我們部署這個有缺陷的版本幾分鐘後，延遲率會以固定的速率增加。
更好的解決方案
為了創建一個2層的channel系統，我們決定使用一個通用模式，一個用來插入作業，一個用來控制作業隊列上同時運行的工作協程。
我們的想法是將並行上傳穩定在一個可持續的速率，這不會削弱機器的性能，也不會產生到S3的連接錯誤。所以我們選擇創建了一個Job / Worker模式。對於熟悉Java，C＃等的人來說，想像一下如何以golang的方式用channel來實現一個worker執行緒池。
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請求句柄，創建一個帶有負載的Job結構體實例，並將其發送到JobQueue channel中以供workers獲取。
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伺服器初始化期間，我們創建了一個Dispatcher並調用Run（）來創建works pool並開始偵聽即將出現在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) } } }注意，我們提供了添加到works pool 中的works 最大數量。既然我們在工程中使用了帶有容器化Go環境的Amazon Elasticbeanstalk，我們就會嘗試一直遵循12-factor方法論來在生產環境中配置我們的系統。因此我們會從環境變數中讀取這些值。這樣我們就可以控制JobQueue的work數量，調整這些值後可以快速生效而無需重新部署集群。
var ( MaxWorker = os.Getenv("MAX_WORKERS") MaxQueue = os.Getenv("MAX_QUEUE") )實時結果
在我們部署新程式碼之後，我們立即看到延遲率下降到了很小的數值，並且處理請求的能力急劇上升。
在我們的Elastic負載均衡完全預熱幾分鐘之後，我們看到我們的ElasticBeanstalk應用程式每分鐘處理了近100萬個請求。並且在早上的幾個小時，請求流量飆升到了每分百萬之上。
伺服器的使用數量從100台下降到了大概20台。
在我們正確配置了群集和自動擴展功能之後，實例數量降到了4x EC2 c4.Large（沒看懂，大概是這個意思），自動縮放配置好之後，只有CPU使用率超過90%並且維持5分鐘，才會產生一個新的實例。
結論
我信奉簡單致勝。我們原本設計了一個使用大量隊列和後台wokers並且部署複雜的系統，但我們決定使用Elasticbeanstalk的自動擴展能力以及Golang為我們提供開箱即用的高效和簡單的並發方法。
你總能為你的工作找到正確的工具。有時候在你的ruby系統中需要一個強大的web處理器，請考慮一下Ruby之外的生態系統，你可以獲得更簡單但更強大的替代解決方案。
