Ограничение количества одновременно работающих горутин при массовой отправке запросов

Условие

Требуется написать механизм, который будет контролировать количество одновременно исполняющихся горутин при отправке большого набора запросов (например, 10 000 штук). Все запросы обязаны быть выполнены, однако в любой момент времени не должно быть более N активных горутин. Обработка ошибок в данном случае не требуется.

package main

import (
  "context"
  "fmt"
  "strconv"
  "sync"

  "github.com/sirupsen/logrus"
)

type Request struct {
  Payload string
}

type Client interface {
  SendRequest(ctx context.Context, request Request) error
}

type client struct{}

func (c client) SendRequest(ctx context.Context, request Request) error {
  fmt.Println("send request", request.Payload)
  return nil
}

func main() {
  ctx := context.Background()
  c := client{}
  requests := make([]Request, 100)
  for i := 0; i < 100; i++ {
    requests[i] = Request{Payload: strconv.Itoa(i)}
  }

  log := logrus.New()
  makeBatchApiCalls(ctx, c, log, requests)
}

func makeBatchApiCalls(ctx context.Context, c Client, log *logrus.Logger, requests []Request) {
  wg := sync.WaitGroup{}
  for _, r := range requests {
    r := r
    wg.Add(1)
    go func() {
      defer wg.Done()
      err := c.SendRequest(ctx, r)
      if err != nil {
        log.WithError(err).Error("send request")
      }
    }()
  }
  wg.Wait()
}

Ваша задача — добавить в функцию makeBatchApiCalls (или создать вспомогательную) логику, которая будет ограничивать количество одновременно запущенных горутин до заданного значения N, при этом гарантируя, что каждый запрос из входного массива будет обработан.

Ответ

Захватывайте слот семафора до запуска goroutine: тогда одновременно существует не больше N рабочих горутин. WaitGroup ждёт выполнения всех вызовов.

Сохраним существующую сигнатуру, например с лимитом 10:

func makeBatchApiCalls(ctx context.Context, c Client,
    log *logrus.Logger, requests []Request) {
    const limit = 10
    slots := make(chan struct{}, limit)
    var wg sync.WaitGroup
    for _, request := range requests {
        slots <- struct{}{}
        wg.Add(1)
        go func(request Request) {
            defer wg.Done()
            defer func() { <-slots }()
            _ = c.SendRequest(ctx, request)
        }(request)
    }
    wg.Wait()
}

В настоящем коде limit можно передать параметром, проверяя N > 0. Ничего не теряется: цикл ждёт свободного слота вместо пропуска запроса. Гарантируется вызов SendRequest для каждого элемента, не успешная доставка при сетевой ошибке. Клиент обязан допускать параллельную работу и завершать запросы; ошибки по условию игнорируются.