用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