[譯]使用Go處理每分鐘百萬請求

這篇文章在medium上很火，做者以實際案例來分析，講得很好。

咱們常常據說使用Go的goroutine和channel很容易實現高併發，那是否是所有代碼都放在goroutine中運行就能夠實現高併發程序了呢？很顯然並非。這篇文章將教你們如何一步一步寫出一個簡單的， 高併發的Go程序。

正文

我在幾家不一樣的公司從事反垃圾郵件，防病毒和反惡意軟件的工做超過15年，如今我知道這些系統最終會由於咱們要天天處理大量數據而變得愈來愈複雜。

目前，我是smsjunk.com的CEO和 KnowBe4的首席架構師，他們都是網絡安全行業的公司。

有趣的是，在過去的10年裏，做爲一名軟件工程師，我參與過的全部Web後端開發大部分都是使用RubyonRails完成的。不要誤會個人意思，我喜歡 RubyonRails，我相信這是一個了不得的生態，可是過了一段時間，你開始以 Ruby的方式思考和設計系統，忘了如何高效和本來能夠利用多線程、並行、快速執行和小的內存消耗來簡化軟件架構。多年來，我是一名C/C++，Delphi和 C＃開發人員，並且我剛開始意識到如何正確的使用工具進行工做可能會有多複雜。

我對互聯網中那些語言和框架戰爭並不太感興趣，好比哪門語言更好，哪一個框架更快。 我始終相信效率，生產力和代碼可維護性主要取決於如何簡單的構建解決方案。

問題

在處理咱們的匿名監測和分析系統時，咱們的目標是可以處理來自數百萬端點的大量POST請求。Web處理程序將收到一個JSON文檔，該文檔可能包含須要寫入 AmazonS3的多個有效內容的集合，以便咱們的 map-reduce系統稍後對這些數據進行操做。

傳統上，咱們會考慮建立一個工做層架構，利用諸如如下的技術棧：

• Sidekiq
• Resque
• DelayedJob
• ElasticbeanstalkWorkerTier
• RabbitMQ
• ...

並搭建2個不一樣的集羣，一個用於web前端，一個用於worker，所以咱們能夠隨意擴容機器來處理即將到來的請求。

從一開始，咱們的團隊就知道咱們能夠在Go中這樣作，由於在討論階段咱們看到這多是一個很是大流量的系統。我一直在使用Go，大約快2年時間了，並且咱們也使用Go開發了一些系統，可是沒有一個系統的流量可以達到這個數量級。咱們首先建立了幾個struct來定義咱們經過POST調用接收到的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處理程序進行了很是簡單粗暴的實現，將每一個請求直接放到新的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)
}
複製代碼

對於通常的併發量，這實際上是可行的，但這很快就證實不能適用於高併發場景。咱們可能有更多的請求，當咱們將第一個版本部署到生產環境時，咱們開始看到的數量級並非如此，咱們低估了併發量。

上述的方法有幾個問題。沒有辦法控制正在工做的goroutine的數量。並且，因爲咱們每分鐘有100萬個POST請求，因此係統很快就崩潰了。

重來

咱們須要找到另外一種的方法。從一開始咱們就開始討論如何讓請求處理程序的生命週期儘量的短，並在後臺產生處理。固然，這是在 RubyonRails必需要作的事情，不然，無論你是使用puma，unicorn仍是 passenger，你的全部的可用的web worker都將阻塞。

那麼咱們就須要利用常見的解決方案來完成這項工做，好比Resque，Sidekiq， SQS等。固然還有其餘工具，由於有不少方法能夠實現。

所以，咱們第二次改進是建立一個buffer channel，咱們能夠將一些做業請求扔進隊列並將它們上傳到S3，因爲咱們能夠控制隊列的最大長度，而且有足夠的RAM來排隊處理內存中的做業，所以咱們認爲只要在通道隊列中緩衝做業就好了。

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
    }
    ...
}
複製代碼

而後，爲了將任務從buffer channel中取出並處理它們，咱們正在使用這樣的方式:

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

說實話，我不知道咱們在想什麼，這確定是一個難熬的夜晚。這種方法並無給咱們帶來什麼提高，咱們用一個緩衝的隊列替換了有缺陷的併發，也只是推遲了問題的產生時間而已。咱們的同步處理器每次只向S3上傳一個有效載荷，因爲傳入請求的速率遠遠大於單個處理器上傳到S3的能力，所以咱們的buffer channel迅速達到極限，隊列已經阻塞而且沒法再往裏邊添加做業。

咱們只是簡單的繞過了這個問題，最終致使咱們的系統徹底崩潰。在咱們部署這個有缺陷的版本後，咱們的延遲持續的升高。

更好的解決方案

咱們決定在Go channel上使用一個通用模式來建立一個 2-tier(雙重)channel系統，一個用來處理排隊的job，一個用來控制有多少worker在 JobQueue上併發工做。

這個想法是將上傳到S3的並行速度提升到一個可持續的速度，同時不會形成機器癱瘓，也不會引起S3的鏈接錯誤。

因此咱們選擇建立一個 Job/Worker模式。對於那些熟悉Java，C＃等的人來講，能夠將其視爲Golang使用channel來實現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請求處理程序以建立具備有效負載的Job struct，並將其發送到 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服務器初始化期間，咱們建立一個Dispatcher並調用Run()來建立worker池並開始監聽JobQueue中出現的Job。

dispatcher := NewDispatcher(MaxWorker) 
dispatcher.Run()
複製代碼

如下是咱們調度程序實現的代碼：

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池中（就是上面的 WorkerPoolChannel）。因爲咱們已經將Amazon Elasticbeanstalk用於Docker化的Go項目，而且咱們始終嘗試遵循12要素方法來配置生產中的系統，所以咱們從環境變量中讀取這些值，這樣咱們就能夠快速調整這些值以控制工做隊列的數量和最大規模，而不須要從新部署集羣。

var ( 
  MaxWorker = os.Getenv("MAX_WORKERS") 
  MaxQueue  = os.Getenv("MAX_QUEUE") 
)
複製代碼

在咱們發佈了這個版本以後，咱們當即看到咱們的全部的請求延遲都降低到了一個很低的數字，咱們處理請求的效率大大提高。

在咱們的彈性負載均衡器徹底熱身以後的幾分鐘，咱們看到咱們的ElasticBeanstalk應用程序每分鐘提供近100萬次請求。一般在早晨的幾個小時裏，流量高峯會超過每分鐘100萬個請求。

咱們部署了新的代碼，服務器的數量從100臺減小到大約20臺。

在恰當地配置了集羣和自動縮放設置之後，咱們在生成環境用4臺EC2 c4就能完成工做了。若是CPU在連續5分鐘內超過90%，彈性自動縮放系統就自動擴容一個新的實例。

結論

簡單老是個人制勝法寶。咱們能夠設計一個擁有多隊列，多後臺進程和難以部署的複雜系統，可是相反咱們決定利用Elasticbeanstalk的自動縮放和高效簡單的方式去併發，Go語言很好的提供了這些功能。

經驗告訴咱們，用最合適的工具去完成工做。有時，當你的 RubyonRails系統須要實現一個很是強大的處理程序時，能夠考慮在 Ruby生態系統以外尋找更簡單且更強大的替代解決方案。

做者：MarcioCastilho

原文：medium.com/smsjunk/han…