Use multiple go routines to pull from a channel

Question

I have a channel that receives slices of maps with this code:

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

The API that I am submitting these records to accepts batches of up to 150. In order to speed this up I would like to spin up 4 go routines to concurrently process the records in the channel. Once the records hit the channel, it doesnt matter what order they get processed.

Currently I made the following update to the above code that runs singularly to process it:

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++
    }
}

How would I modify this to have 4 go routines pull from the channel and send these requests? Or, is this even possible/have I misunderstood the capabilities of go routines?

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

Playground example that uses int's and 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
}