Go routine:Making concurrent API requests

Question

I am trying to understand channels and goroutines and tried to write a goroutine for making concurrent API requests to the server

But when I am running the code using a goroutine, it seems like it is taking the same time as it does without a goroutine.

func sendUser(user string, ch chan<- string)  {
    resp,err := http.get("URL"/user)
    //do the processing and get resp=string
    ch <- resp
}


func AsyncHTTP(users []string) ([]string, error) {
    ch := make(chan string)
    var responses []string
    var user string

    for _ , user = range users {
        go sendUser(user, ch)

        for {
            select {
            case r := <-ch:
                if r.err != nil {
                    fmt.Println(r.err)
                }
                responses = append(responses, r)
                **//Is there a better way to show that the processing of response is complete**?
                if len(responses) == len(users) { 
                    return responses, nil
                }
            case <-time.After(50 * time.Millisecond):
                fmt.Printf(".")
            }
        }
    }
    return responses, nil
}

Questions:

Even though I am using a goroutine, request completion time is same as it is without goroutines? Is there anything I am doing wrong with goroutines?
For telling the job not to wait anymore here I am using:
```
if len(responses) == len(users)
```
Is there a better way to show that the processing of response is complete and tell ch not to wait anymore?
What is wait.Syncgroup? How can I use it in my goroutine?

user_2649611 · Answer

For bounded parallelism / rate limiting, we can take a look an example at https://blog.golang.org/pipelines#TOC_9.

Basically the steps are:

Stream inputs / params / args used to call the API, to an input channel.
Run N worker goroutines, each consuming the same (shared) input channel. Get the args from input channel, call the API, send the result into a result channel.
Consume the result channel, return early if there's error.

sync.WaitGroup is used to wait for all worker goroutines to complete (after the input channel is exhausted).

Below is code example of it (you can run it right away, try changing NUM_PARALLEL to different number of parallelism). Change BASE_URL to your base url.

package main

import (
    "fmt"
    "io"
    "net/http"
    "strconv"
    "sync"
    "time"
)

// placeholder url. Change it to your base url.
const BASE_URL = "https://jsonplaceholder.typicode.com/posts/"

// number of parallelism
const NUM_PARALLEL = 20

// Stream inputs to input channel
func streamInputs(done <-chan struct{}, inputs []string) <-chan string {
    inputCh := make(chan string)
    go func() {
        defer close(inputCh)
        for _, input := range inputs {
            select {
            case inputCh <- input:
            case <-done:
                // in case done is closed prematurely (because error midway),
                // finish the loop (closing input channel)
                break
            }
        }
    }()
    return inputCh
}

// Normal function for HTTP call, no knowledge of goroutine/channels
func sendUser(user string) (string, error) {
    url := BASE_URL + user
    resp, err := http.Get(url)
    if err != nil {
        return "", err
    }
    defer resp.Body.Close()

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return "", err
    }

    bodyStr := string(body)
    return bodyStr, nil
}

// Wrapper for sendUser return value, used as result channel type
type result struct {
    bodyStr string
    err     error
}

func AsyncHTTP(users []string) ([]string, error) {
    done := make(chan struct{})
    defer close(done)

    inputCh := streamInputs(done, users)

    var wg sync.WaitGroup
    // bulk add goroutine counter at the start
    wg.Add(NUM_PARALLEL)

    resultCh := make(chan result)

    for i := 0; i < NUM_PARALLEL; i++ {
        // spawn N worker goroutines, each is consuming a shared input channel.
        go func() {
            for input := range inputCh {
                bodyStr, err := sendUser(input)
                resultCh <- result{bodyStr, err}
            }
            wg.Done()
        }()
    }

    // Wait all worker goroutines to finish. Happens if there's no error (no early return)
    go func() {
        wg.Wait()
        close(resultCh)
    }()

    results := []string{}
    for result := range resultCh {
        if result.err != nil {
            // return early. done channel is closed, thus input channel is also closed.
            // all worker goroutines stop working (because input channel is closed)
            return nil, result.err
        }
        results = append(results, result.bodyStr)
    }

    return results, nil
}

func main() {
    // populate users param
    users := []string{}
    for i := 1; i <= 100; i++ {
        users = append(users, strconv.Itoa(i))
    }

    start := time.Now()

    results, err := AsyncHTTP(users)
    if err != nil {
        fmt.Println(err)
        return
    }

    for _, result := range results {
        fmt.Println(result)
    }

    fmt.Println("finished in ", time.Since(start))
}

Go routine:Making concurrent API requests

2 Answers