使用 Go 语言每分钟处理 1 百万请求

type PayloadCollection struct {
    WindowsVersion  string    `json:"version"`
    Token           string    `json:"token"`
    Payloads        []Payload `json:"data"`
}

type Payload struct {
    // [编写]
}

func (p *Payload) UploadToS3() error {
    //  storageFolder 方法避免时间戳相同时产生的同名 key 冲突 
    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
    }

    // 我们传到 S3 存储块中的所有东西都应该标记为 'private'
    var acl = s3.Private
    var contentType = "application/octet-stream"

    return bucket.PutReader(storage_path, b, int64(b.Len()), contentType, acl, s3.Options{})
}Copy

func payloadHandler(w http.ResponseWriter, r *http.Request) {

    if r.Method != "POST" {
        w.WriteHeader(http.StatusMethodNotAllowed)
        return
    }

    // 将包体读入到一个字符串进行 json 解析
    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
    }

    // 遍历每个负载，将其条目逐一队列化，以期把它们发到 S3
    for _, payload := range content.Payloads {
        go payload.UploadToS3()   // <----- 千万别这么干
    }

    w.WriteHeader(http.StatusOK)
}Copy

var Queue chan Payload

func init() {
    Queue = make(chan Payload, MAX_QUEUE)
}

func payloadHandler(w http.ResponseWriter, r *http.Request) {
    ...
    // 遍历每个有效载荷，逐个排队之后上传 S3
    for _, payload := range content.Payloads {
        Queue <- payload
    }
    ...
}Copy

func StartProcessor() {
    for {
        select {
        case job := <-Queue:
            job.payload.UploadToS3()  // <-- 仍不够好
        }
    }
}Copy

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

// Job 指代要运行的任务
type Job struct {
    Payload Payload
}

// 一个可以发送工作请求的缓冲 channel
var JobQueue chan Job

// Worker 表示执行任务的 worker
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 方法开启一个 worker 循环，监听退出 channel，可按需停止这个循环
func (w Worker) Start() {
    go func() {
        for {
            // 将当前的 worker 注册到 worker 队列中
            w.WorkerPool <- w.JobChannel

            select {
            case job := <-w.JobChannel:
                // 此时我们接收到一个工作请求
                if err := job.Payload.UploadToS3(); err != nil {
                    log.Errorf("Error uploading to S3: %s", err.Error())
                }

            case <-w.quit:
                // 此处接收一个停止信号
                return
            }
        }
    }()
}

// Stop 方法控制 worker 停止监听工作请求
func (w Worker) Stop() {
    go func() {
        w.quit <- true
    }()
}Copy

func payloadHandler(w http.ResponseWriter, r *http.Request) {

    if r.Method != "POST" {
        w.WriteHeader(http.StatusMethodNotAllowed)
        return
    }

    // 把包体读入到一个字符串中，进行 json 解析
    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
    }

    // 遍历载荷，逐个队列化用以上传到 S3
    for _, payload := range content.Payloads {

        // 创建一个带有载荷的任务（ Job 实例 ）
        work := Job{Payload: payload}

        // 然后把它放到队列中
        JobQueue <- work
    }

    w.WriteHeader(http.StatusOK)
}Copy

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

type Dispatcher struct {
    // 注册到 dispatcher 的 worker channel 池
    WorkerPool chan chan Job
}

func NewDispatcher(maxWorkers int) *Dispatcher {
    pool := make(chan chan Job, maxWorkers)
    return &Dispatcher{WorkerPool: pool}
}

func (d *Dispatcher) Run() {
    // 开始运行 n 个 worker
    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:
            // 收到任务请求
            go func(job Job) {
                // 尝试获取一个可用的 worker job channel
                // 阻塞直到有可用的 worker
                jobChannel := <-d.WorkerPool

                // 分发任务到 worker job channel 中
                jobChannel <- job
            }(job)
        }
    }
}Copy

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