使用golang每分鐘處理百萬請求

　　問題
　　
　　從原生方法到Go協程
　　
　　再試一次
　　
　　更好的解決方案
　　
　　實時結果
　　
　　結論
　　
　　正文
　　
　　[譯]使用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 http://www.senta7.net/content/?729.html:= 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(www.hengxuangyul.com"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 www.xcdeyiju.com{
　　
　　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 <www.zhonxinyule.com- w.JobChannel
　　
　　select {
　　
　　case job :www.yacuangyl.com= <-w.JobChannel:
　　
　　// we have received a work request.
　　
　　if err := job.Payload.UploadToS3(www.jujinyule.com); 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以外的生態系統，你能夠得到更簡單但更強大的替代解決方案。