Golang中怎么處理每分鐘百萬請求

Golang中怎么處理每分鐘百萬請求，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

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

• Sidekiq

• Resque

• DelayedJob

• ElasticbeanstalkWorkerTier

• RabbitMQ

• ...

并搭建2個不同的集群，一個用于web前端，一個用于worker，因此我們可以隨意擴容機器來處理即將到來的請求。

從一開始，我們的團隊就知道我們可以在Go中這樣做，因為在討論階段我們看到這可能是一個非常大流量的系統(tǒng)。我一直在使用Go，大約快2年時間了，而且我們也使用Go開發(fā)了一些系統(tǒng)，但是沒有一個系統(tǒng)的流量能夠達到這個數(shù)量級。我們首先創(chuàng)建了幾個struct來定義我們通過POST調(diào)用接收到的Web請求，并將其上傳到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{}) }

Naive的做法-硬核使用Goroutine

最初，我們對POST處理程序進行了非常簡單粗暴的實現(xiàn)，將每個請求直接放到新的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()   // <----- DON'T DO THIS     }      w.WriteHeader(http.StatusOK) }

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

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

重來

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

那么我們就需要利用常見的解決方案來完成這項工作，比如Resque，Sidekiq， SQS等。當(dāng)然還有其他工具，因為有很多方法可以實現(xiàn)。

因此，我們第二次改進是創(chuàng)建一個buffer  channel，我們可以將一些作業(yè)請求扔進隊列并將它們上傳到S3，由于我們可以控制隊列的***長度，并且有足夠的RAM來排隊處理內(nèi)存中的作業(yè)，因此我們認為只要在通道隊列中緩沖作業(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         }     } }

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

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

更好的解決方案

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

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

所以我們選擇創(chuàng)建一個  Job/Worker模式。對于那些熟悉Java，C#等的人來說，可以將其視為Golang使用channel來實現(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請求處理程序以創(chuàng)建具有有效負載的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)建一個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)         }     } }

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

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

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

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

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

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

看完上述內(nèi)容，你們掌握Golang中怎么處理每分鐘百萬請求的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！