使用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生態系統之外尋找更簡單且更強大的替代解決方案。
