天天看點

Go 采用 goroutine 和 channel 實作工作池

假設有一組任務需要異步處理且量很大,那我們需要同時開啟多個 worker 以保證任務的處理速度而不會堵塞任務。其他語言,可能會需要開啟多程序來完成,多程序的控制、IO 消耗等會是個需要注意的問題,而這些 Go 都能幫我們很輕易的解決。

大緻的實作要點和流程:

  • 建立2個信道,messages 用于傳送任務消息,result 用于接收消息處理結果
  • 建立3個 Worker 協程,用于接收和處理來自 messages 信道的任務消息,并将處理結果通過信道 result 傳回
  • 通過信道 messages 釋出10條任務
  • 通過信道 result 接收任務處理結果

示例代碼:

package main

import (
    "fmt"
    "strconv"
    "math/rand"
    "time"
)

type Message struct {
    Id   int
    Name string
}

func main() {
    messages := make(chan Message, 100)
    result := make(chan error, 100)

    // 建立任務處理Worker
    for i := 0; i < 3; i ++ {
        go worker(i, messages, result)
    }

    total := 0
    // 釋出任務
    for k := 1; k <= 10; k ++ {
        messages <- Message{Id: k, Name: "job" + strconv.Itoa(k)}
        total += 1
    }

    close(messages)

    // 接收任務處理結果
    for j := 1; j <= total; j ++ {
        res := <-result
        if res != nil {
            fmt.Println(res.Error())
        }
    }

    close(result)
}

func worker(worker int, msg <-chan Message, result chan<- error) {
    // 從通道 chan Message 中監聽&接收新的任務
    for job := range msg {
        fmt.Println("worker:", worker, "msg: ", job.Id, ":", job.Name)

        // 模拟任務執行時間
        time.Sleep(time.Second * time.Duration(RandInt(1, 3)))

        // 通過通道傳回執行結果
        result <- nil
    }
}

func RandInt(min, max int) int {
    rand.Seed(time.Now().UnixNano())
    return min + rand.Intn(max-min+1)
}           
原文位址: https://shockerli.net/post/golang-worker-pools/