Golang中怎么处理每分钟百万请求

Golang中怎么处理每分钟百万请求，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。

成都创新互联公司是一家业务范围包括IDC托管业务,虚拟空间、主机租用、主机托管，四川、重庆、广东电信服务器租用,成都天府联通服务器托管，成都网通服务器托管,成都服务器租用,业务范围遍及中国大陆、港澳台以及欧美等多个国家及地区的互联网数据服务公司。

传统上，我们会考虑创建一个工作层架构，利用诸如以下的技术栈：

• 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%，弹性自动缩放系统就自动扩容一个新的实例。

看完上述内容，你们掌握Golang中怎么处理每分钟百万请求的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注创新互联行业资讯频道，感谢各位的阅读！