使用多个 goroutine 从通道中拉取

Question

我有一个接收带有此代码的地图切片的频道：

func submitRecords(records []map[string]string) {
    batch := []map[string]string{}
    ch := make(chan []map[string]string)
    batchCt := 1
    go func() {
        for _, v := range records {
            batch = append(batch, v)
            if len(batch) == 150 {
                ch <- batch
                batch = nil
            }
        }
        close(ch)
    }()
}

我提交这些记录的 API 接受最多 150 个批次。为了加快速度，我想启动 4 个 go 例程来同时处理通道中的记录。一旦记录进入频道，它们被处理的顺序就无关紧要了。

目前我对上面单独运行的代码进行了以下更新：

func submitRecords(records []map[string]string) {
    batch := []map[string]string{}
    ch := make(chan []map[string]string)
    batchCt := 1
    go func() {
        for _, v := range records {
            batch = append(batch, v)
            if len(batch) == 150 {
                ch <- batch
                batch = nil
            }
        }
        close(ch)
    }()

    for b := range ch {
        str, _ := json.Marshal(b)
        fmt.Printf("Sending batch at line %d\n", (batchCt * 150))

        payload := strings.NewReader(string(str))
        client := &http.Client{}
        req, err := http.NewRequest(method, url, payload)
        if err != nil {
            fmt.Println(err)
        }

        login, _ := os.LookupEnv("Login")
        password, _ := os.LookupEnv("Password")
        req.Header.Add("user_name", login)
        req.Header.Add("password", password)
        req.Header.Add("Content-Type", "application/json")

        res, err := client.Do(req)
        if err != nil {
            fmt.Println(err)
        }
        batchCt++
    }
}

我将如何修改它以从通道中提取 4 个 go 例程并发送这些请求？或者，这是否可能/我是否误解了 goroutine 的功能？

Answer 1

func process(ch chan []map[string]string) {
    for b := range ch {
        str, _ := json.Marshal(b)

        // This wont work, or has to be included in the payload from the channel
        // fmt.Printf("Sending batch at line %d\n", (batchCt * 150))

        payload := strings.NewReader(string(str))
        client := &http.Client{}
        req, err := http.NewRequest(method, url, payload)
        if err != nil {
            fmt.Println(err)
        }

        login, _ := os.LookupEnv("Login")
        password, _ := os.LookupEnv("Password")
        req.Header.Add("user_name", login)
        req.Header.Add("password", password)
        req.Header.Add("Content-Type", "application/json")

        res, err := client.Do(req)
        if err != nil {
            fmt.Println(err)
        }
        // batchCt++
    }
    done <- 1
}

func submitRecords(records []map[string]string) {
    batch := []map[string]string{}
    ch := make(chan []map[string]string)
    
    go process(ch)
    go process(ch)
    go process(ch)
    go process(ch)

    // batchCt := 1
    for _, v := range records {
        batch = append(batch, v)
        if len(batch) == 150 {
            ch <- batch
            batch = []map[string]string{}
        }
    }
    // Send the last not to size batch
    ch <- batch
    close(ch)
}

使用 int 和 sleep https://play.golang.org/p/q5bUhXt9aUn 的 Playground 示例

package main

import (
    "fmt"
    "time"
)

func process(processor int, ch chan []int, done chan int) {
    for batch := range ch {
        // Do something.. sleep or http requests will let the other workers work as well
        time.Sleep(time.Duration(len(batch)) * time.Millisecond)
        fmt.Println(processor, batch)
    }
    done <- 1
}

const batchSize = 3

func main() {
    records := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13}
    ch := make(chan []int)
    done := make(chan int)

    go process(1, ch, done)
    go process(2, ch, done)
    go process(3, ch, done)
    go process(4, ch, done)

    batch := make([]int, 0, batchSize)
    for _, v := range records {
        batch = append(batch, v)
        if len(batch) == batchSize {
            ch <- batch
            batch = make([]int, 0, batchSize)
        }
    }
    ch <- batch
    close(ch)

    <-done
    <-done
    <-done
    <-done
}