這篇文章給大家介紹Golang中怎么應付百萬級請求，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

創(chuàng)新互聯(lián)專注于揭陽網站建設服務及定制，我們擁有豐富的企業(yè)做網站經驗。 熱誠為您提供揭陽營銷型網站建設，揭陽網站制作、揭陽網頁設計、揭陽網站官網定制、小程序開發(fā)服務，打造揭陽網絡公司原創(chuàng)品牌,更為您提供揭陽網站排名全網營銷落地服務。

• 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 runtines

最開始的時候我們非常天真地實現(xiàn)一個POST的鉤子方法如下，只是簡單地將每個請求體的上傳動作放到Go rutinues中讓他們并行執(zhí)行：

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

在中等規(guī)模的負載情況下，這種方法對大部分人都是沒有問題的，但在應對更大規(guī)模的請求量時候，我們很快就招架不住了。當我們把這個版本的代碼部署到生產環(huán)境以后，我們期待能有大量的請求進來但實際還不能達到百萬級別的數(shù)量級。我們完全低估了這個系統(tǒng)要處理的流量數(shù)。

但不管怎么說上面的方法都是欠妥的。因為它沒有任何方法讓我們去控制Go  runtinues啟動的數(shù)量。所以當我們的系統(tǒng)在面對每分鐘百萬級POST請求的時候很快就垮掉了。

再戰(zhàn)

我們需要找到另外的方法。在一開始我們就在討論如何讓我們的請求處理程序的生命周期盡可能地縮短以及上傳到S3的操作能在后臺或者異步運行。當然，在Ruby on  Rails里面你必須這么做，否則你將會阻塞到所有其他的網絡請求處理程序。無論您使用的是美洲獅，獨角獸還是過路人(請不要參與JRuby討論)。然后我們想到使用消息隊列這種比較常見的方法來處理來達到我們的目的，例如Resque,  Sidekiq, SQS等等，還有數(shù)不清的工具因為實在有太多方法來實現(xiàn)這個功能。

所以在第二次迭代的時候，我們需要創(chuàng)建一個緩沖隊列，我們會將任務放入隊列里面然后再一個個地上傳到S3上，但由于我們希望達到能夠控制這個隊列的最大容量的目的，并且我們有足夠的RAM來允許我們將請求體儲存到內存當中，所以我們認為直接使用了Go提供的channel，然后將我們的請求直接入隊到channel中處理就可以了。

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

我們會從channel中獲取任務并且執(zhí)行他們的上傳操作

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

但說句老實話，我并不知道這是在干嘛?？隙ㄊ且驗槟菚r已經太晚還有我們已經喝了太多的紅牛。

這個改動并沒有讓我們的困境得到任何改善，我們將并發(fā)任務放到了隊列中執(zhí)行僅僅是看上去好像解決了問題。但是我們的異步程序一次只會上傳一個請求體到S3上面，但是我們的請求數(shù)此時遠遠大于我們上傳到S3的數(shù)量，可想而知我們的緩沖隊列很快就到達了他的極限爆滿了，然后它阻擋了其他網絡請求的入隊操作。

相當于我們僅僅回避了問題，并且讓我們的系統(tǒng)的崩潰時間進入了倒數(shù)。我們這個缺陷的版本發(fā)布以后，整個系統(tǒng)的延遲率在持續(xù)性地每分鐘在上漲。



• 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 

•     }() 

• } 

接下來修改我們網絡請求的鉤子函數(shù)，負責創(chuàng)建一個Job的結構體的實例然后將其放入JobQueue channel中等待worker來獲取執(zhí)行。

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

在我們網絡服務初始化的時候創(chuàng)建一個Dispather并且調用Run()創(chuàng)建一個裝有一定數(shù)量worker的線程池，用來接收和處理來自JobQueue的Job

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

下面是我們Dispather的實現(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)         }     } }

當我們將這個版本發(fā)布到生產環(huán)境以后我們的延遲率馬上有明顯的下降，我們處理請求的能力有一個質的飛躍。

關于Golang中怎么應付百萬級請求就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。

新聞名稱：Golang中怎么應付百萬級請求
URL地址：http://jinyejixie.com/article28/pdcscp.html

成都網站建設公司_創(chuàng)新互聯(lián)，為您提供用戶體驗、品牌網站設計、網站策劃、網站導航、建站公司、品牌網站制作

廣告

聲明：本網站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)