go - 使用多个go例程从 channel 中提取

Question

我有一个使用以下代码接收 map slice 的 channel :

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

我提交这些记录以接受最多150个批次的API。为了加快速度，我想启动4个go例程以同时处理 channel 中的记录。记录到达 channel 后，按什么顺序处理它们都没有关系。
当前，我对上面运行的代码进行了以下更新，以单独处理该代码:

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例程从 channel 中拉出并发送这些请求？或者，这是否有可能/我是否误解了go例程的功能？

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

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
}