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

type Payload struct {
    // ...負(fù)載字段
}

func (p *Payload) UploadToS3() error {
    // storageFolder 方法確保在我們?cè)阪I名中獲得相同時(shí)間戳?xí)r不會(huì)發(fā)生名稱(chēng)沖突
    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
    }

    // 我們發(fā)布到 S3 存儲(chǔ)桶的所有內(nèi)容都應(yīng)標(biāo)記為“私有”
    var acl = s3.Private
    var contentType = "application/octet-stream"

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

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

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

    // 將body讀入字符串進(jìn)行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
    }
    
    // 分別檢查每個(gè)有效負(fù)載和隊(duì)列項(xiàng)目以發(fā)布到 S3
    for _, payload := range content.Payloads {
        go payload.UploadToS3()   // <----- 這是不建議的做法。這里是最開(kāi)始的做法。
    }

    w.WriteHeader(http.StatusOK)
}

var Queue chan Payload

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

func payloadHandler(w http.ResponseWriter, r *http.Request) {
    ...
    // 分別檢查每個(gè)有效負(fù)載和隊(duì)列項(xiàng)目以發(fā)布到 S3
    for _, payload := range content.Payloads {
        Queue <- payload // <----- 這是建議的做法。
    }
    ...
}

func StartProcessor() {
    for {
        select {
        case job := <-Queue:
            job.payload.UploadToS3()  // <-- 這里雖然優(yōu)化了，但還不是最好的。
        }
    }
}

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

// Job 表示要運(yùn)行的作業(yè)
type Job struct {
    Payload Payload
}

// 我們可以在 Job 隊(duì)列上發(fā)送工作請(qǐng)求的緩沖通道。
var JobQueue chan Job

// Worker 代表執(zhí)行作業(yè)的 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 啟動(dòng)循環(huán)監(jiān)聽(tīng)。監(jiān)聽(tīng)退出信號(hào)以防我們需要停止它。
func (w Worker) Start() {
    go func() {
        for {
            // 將當(dāng)前 woker 注冊(cè)到工作隊(duì)列中。
            w.WorkerPool <- w.JobChannel

            select {
            case job := <-w.JobChannel:
                // 接收 work 請(qǐng)求。
                if err := job.Payload.UploadToS3(); err != nil {
                    log.Errorf("Error uploading to S3: %s", err.Error())
                }

            case <-w.quit:
                // 接收一個(gè)退出的信號(hào)。
                return
            }
        }
    }()
}

// 將退出信號(hào)傳遞給 Worker 進(jìn)程以停止處理清理。
func (w Worker) Stop() {
    go func() {
        w.quit <- true
    }()
}

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

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

    // 將body讀入字符串進(jìn)行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
    }

    // 分別檢查每個(gè)有效負(fù)載和隊(duì)列項(xiàng)目以發(fā)布到 S3
    for _, payload := range content.Payloads {

        // 創(chuàng)建一個(gè)有效負(fù)載的job
        work := Job{Payload: payload}

        // 將 work push 到隊(duì)列。
        JobQueue <- work
    }

    w.WriteHeader(http.StatusOK)
}

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

type Dispatcher struct {
    // 通過(guò)調(diào)度器注冊(cè)一個(gè) Worker 通道池
    WorkerPool chan chan Job
}

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

func (d *Dispatcher) Run() {
    // 啟動(dòng)指定數(shù)量的 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:
            // 接收一個(gè) job 請(qǐng)求
            go func(job Job) {
                // 嘗試獲取可用的 worker job 通道
                // 這將阻塞 worker 直到空閑
                jobChannel := <-d.WorkerPool

                // 調(diào)度一個(gè) job 到 worker job 通道
                jobChannel <- job
            }(job)
        }
    }
}

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