用Golang處理每分鐘100萬份請求

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

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

有趣的是，在過去的10年左右，做爲一名軟件工程師，我參與過的全部Web後端開發大部分都是在Ruby on Rails中完成的。不要誤會個人意思，我喜歡Ruby on Rails，我相信這是一個了不得的環境，可是過了一段時間，你開始用ruby的方式思考和設計系統，並且若是你忘記了軟件架構的效率和簡單性-能夠利用多線程，並行化，快速執行和小內存開銷。多年來，我是一名C / C ++，Delphi和C＃開發人員，並且我剛開始意識到使用正確的工具進行工做可能會有多複雜。

我對互聯網老是爭論的語言和框架戰爭並不太感興趣。我相信效率，生產力和代碼可維護性主要取決於您構建解決方案的簡單程度。

問題

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

傳統上，咱們會考慮建立一個工做層架構，利用諸如如下方面的內容：

• Sidekiq
• Resque
• DelayedJob
• Elasticbeanstalk Worker Tier
• 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的作法-使用Go routine

最初，咱們對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) } 複製代碼

對於中等負載，這能夠適用於大多數人，但這很快就證實不能很好地大規模工做。咱們期待着不少請求，可是在咱們將第一個版本部署到生產環境時，咱們開始看到的數量級並非如此。咱們忽視了流量。

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

重來

咱們須要找到一種不一樣的方式。從一開始咱們就開始討論如何保持請求處理程序的生命週期很是短，並在後臺產生處理。固然，這就是Ruby on Rails必需要作的事情，不然，無論你是使用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上使用一個通用模式來建立一個雙層channel系統，一個用來處理排隊的job，一個用來控制有多少worker在JobQueue 上併發工做。

這個想法是將上傳到S3的並行化速度提升到一個可持續的速度，不會形成機器癱瘓，也不會引起S3的鏈接錯誤。 因此咱們選擇建立一個Job / Worker模式。對於那些熟悉Java，C＃等的人來講，能夠將其視爲Golang使用channel來實現Worker Thread-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，並將其發送到JobQueue channel以供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池中（就是上面的WorkerPool channel）。因爲咱們已經將Amazon Elasticbeanstalk用於Docker化的Go項目，而且咱們始終嘗試遵循12因子方法來配置生產中的系統，所以咱們從環境變量中讀取這些值。這樣咱們就能夠控制工做隊列的數量和最大規模，因此咱們能夠快速調整這些值，而不須要從新部署集羣。

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

在咱們部署它以後，咱們當即看到咱們的全部延遲率都降低到微不足道的數字，咱們處理請求的能力急劇上升。

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

只要咱們部署了新代碼，服務器的數量就會從100臺服務器大幅降低到大約20臺服務器。

在咱們正確地配置了咱們的集羣和自動縮放設置後，咱們能夠將它下降到只有4x EC2 c4的配置。大型實例和Elastic Auto-Scaling設置爲在CPU連續5分鐘超過90％時產生一個新實例。

結論

樸素老是在個人書中獲勝。咱們能夠設計一個擁有許多隊列，後臺工做人員和複雜部署的複雜系統，但咱們決定利用Elasticbeanstalk自動擴展的強大功能以及Golang爲咱們提供開箱即用的併發性效率和簡單方法。

並非天天都是隻有4臺機器的集羣，這可能遠不及我如今的MacBook Pro，它可以每分鐘處理100w次的請求。

老是有適合指定需求的工具。有時，當您的Ruby on Rails系統須要一個很是強大的Web處理程序時，請考慮在Ruby生態系統以外尋找更簡單但更強大的替代解決方案。

原文地址：https://medium.com/smsjunk/handling-1-million-requests-per-minute-with-golang-f70ac505fcaa