用Golang处理每分钟百万级请求

时间  2019-11-09
标签 golang 处理 每分钟 百万 请求 栏目 Go 繁體版
原文   原文链接

翻译原文连接 转帖／转载请注明出处html

原文连接@medium.com 发表于2017/08/30git

我在防垃圾邮件，防病毒和防恶意软件领域已经工做了15年，先后在好几个公司任职。我知道这些系统最后都会由于要处理海量的数据而变得很是复杂。github

我如今是smsjunk.com的CEO而且是KnowBe4的首席架构师。这两个公司在网络安全领域都很是活跃。golang

有趣的是，在过去10年里做为一个码农，全部我经历过的网站后台开发用的几乎都是用Ruby on Rails。不要误解，我很喜欢Ruby on Rails而且认为它是一个很是棒的开发环境。每每在一段时间后，你开始以ruby的方式来设计系统。这时你会忘记利用多线程，并行，快速执行（fast executions）和较小的内存开销（small memory overhead），软件的架构会变得简单而高效。不少年来，我一直是C/C++，Delphi和C#的开发者。我开始意识到使用正确的工具，工做会变得简单不少。docker

我对语言和框架并非很热衷。我相信效率，产出和代码的可维护性取决于你如何架构一个简洁地解决方案。编程

问题

在开发咱们的匿名遥测和分析系统时，咱们的目标是可以处理从上百万个端点发来的大量POST请求。HTTP请求处理函数会收到包含不少载荷（payloads）的JSON文档。这些载荷（payloads）须要被写到Amazon S3上，接着由map-reduce系统来处理。json

一般咱们会建立一个worker池架构（worker-tier architecture)。利用以下的一些工具：缓存

而后设置两个集群，一个用做处理HTTP请求，另一个用做workers。这样咱们可以根据处理的后台工做量进行扩容。

从一开始咱们小组就以为应该用Go来实现，由于在讨论阶段咱们估计这可能会是一个处理很是大流量的系统。我已经使用Go语言两年并用它在工做中开发了一些系统，但它们都没有处理过这么大的负载（load）。

咱们首先建立了几个结构来定义HTTP请求的载荷。咱们经过POST请求接收这些载荷，而后用一个函数上传到S3 bucket。

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{})
}

简单地使用Goroutines

一开始咱们用了最简单的方法来实现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的数量。由于咱们每分钟收到了一百万个POST请求，上面的代码很快就奔溃了。

再次尝试

咱们须要一个不一样的解决方案。在一开始，咱们就讨论到须要把HTTP请求处理函数写的简洁，而后把处理工做转移到后台。固然，这是你在Ruby on Rails世界里必须作的，不然你会阻塞全部worker的工做（例如puma，unicorn，passenger等等，咱们这里就不继续讨论JRuby了）。咱们须要用到Resque，Sidekiq，SQS等经常使用的解决方案。这个列表能够很长，由于有许多方法来完成这项任务。

第二个版本是建立带缓冲的channel。这样咱们能够把工做任务放到队列里而后再上传到S3。由于能够控制队列的长度而且有充足的内存，咱们以为把工做任务缓存在channel队列里应该没有问题。

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
    }
    ...
}

而后咱们须要从队列里提取工做任务并进行处理。代码下图所示：

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

坦白的说，我不知道咱们当时在想什么。这确定是熬夜喝红牛的结果。这个方法并无给咱们带来任何帮助。队列仅仅是将问题延后了。咱们的处理函数（processor）一次仅上传一个载荷（payload），而接收请求的速率比一个处理函数上传S3的能力大太多了，带缓冲的channel很快就到达了它的极限。从而阻塞了HTTP请求处理函数往队列里添加更多的工做任务。

咱们仅仅是延缓了问题的触发。系统在倒计时，最后仍是崩溃了。在这个有问题的版本被部署以后，系统的延迟以恒定速度在不停地增加。

0_latency.png

更好的解决办法

咱们决定使用Go channel的经常使用编程模式。使用一个两级channel系统，一个用来存听任务队列，另外一个用来控制处理任务队列的并发量。

这里的想法是根据一个可持续的速率将S3上传并行化。这个速率不会使机器变慢或者致使S3的链接错误。咱们选择了一个Job／Worker模式。若是大家对Java，C#等语言熟悉的话，能够把它想象成是Go语言用channel来实现的工做线程池。

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
    }()
}

咱们修改了HTTP请求处理函数来建立一个含有载荷（payload）的Job结构，而后将它送到一个叫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)
}

在初始化服务的时候，咱们建立了一个Dispatcher而且调用了Run()函数来建立worker池。这些worker会监听JobQueue上是否有新的任务并进行处理。

dispatcher := NewDispatcher(MaxWorker)
dispatcher.Run()

下面是咱们的dispatcher实现代码：

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加入到worker池里。由于咱们已经在docker化的Go环境里使用了Amazon的Elasticbeanstalk而且严格按照12-factor方法来配置咱们的生产环境，这些参数值能够从环境变量里得到。咱们能够方便地控制worker数目和任务队列的长度。咱们能够快速地调整这些值而不须要从新部署整个集群。

var (
  MaxWorker = os.Getenv("MAX_WORKERS")
  MaxQueue  = os.Getenv("MAX_QUEUE")
)

部署了新版本以后，咱们看到系统延迟一会儿就降到了能够忽略的量级。同时处理请求的能力也大幅攀升。

1_latency.png

在Elastic Load Balancers热身后几分钟，咱们看到Elasticbeanstalk应用开始处理将近每分钟一百万个请求。咱们的流量一般在早上的时候会攀升至超过每分钟一百万个请求。同时，咱们也将服务器的数目从100台缩减到了20台。

2_host.png

经过合理地配置集群和auto-scaling，咱们可以作到只配置4台EC2 c4.Large实例。而后当CPU使用率持续5分钟在90%以上时用Elastic Auto-Scaling来建立新的实例。

3_util.png

结束语

对我来讲简洁（simplicity）是第一位的。咱们能够利用无数队列，不少后台worker以及复杂的部署来设计一个复杂系统，最终咱们仍是使用了Elasticbeanstalk auto-scaling的强大功能和Go语言提供的应对并发的简单方法。用仅仅4台机器（可能还不如个人MacBook Pro强大）来处理每分钟一百万次POST请求对Amazon S3进行写操做。

每项任务都有对应的正确工具。当你的Ruby on Rails系统须要一个很强大的HTTP请求处理器，能够尝试看看ruby生态系统之外的其它更强大的选项。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程