Представим сервис, который получает пачку изображений и строит миниатюры. Обработка независима, но запускать сотню операций одновременно нельзя: закончатся память, файловые дескрипторы или терпение соседнего API.

Нам нужен ограниченный параллелизм. В Go его естественно выразить фиксированным числом горутин, читающих общий канал. Важная часть начинается не в цикле обработки, а вокруг него: как задача попадает в очередь, что происходит при отмене и кто закрывает канал.

Минимальный рабочий пул

type Job struct {
    ID   string
    Path string
}

func Run(ctx context.Context, workers int, jobs <-chan Job) error {
    group, ctx := errgroup.WithContext(ctx)

    for n := 0; n < workers; n++ {
        group.Go(func() error {
            for {
                select {
                case <-ctx.Done():
                    return ctx.Err()
                case job, ok := <-jobs:
                    if !ok {
                        return nil
                    }
                    if err := makeThumbnail(ctx, job); err != nil {
                        return fmt.Errorf("job %s: %w", job.ID, err)
                    }
                }
            }
        })
    }

    return group.Wait()
}

Количество воркеров — явный предел параллелизма. Первый сбой отменяет общий контекст, остальные обработчики получают сигнал остановки, а вызывающий код — ошибку с идентификатором задания. Эту политику легко прочитать прямо из функции.

Буфер — это не бесплатная скорость

Буферизированный канал отделяет производителя от потребителей на коротком всплеске. Но большой буфер также прячет перегрузку и удерживает память. Если воркеры стабильно медленнее источника, очередь всё равно когда-нибудь заполнится.

Я обычно начинаю с буфера в один-два раза больше числа воркеров. Когда канал заполнен, отправитель ждёт. Это и есть обратное давление: система перестаёт принимать работу быстрее, чем может её выполнить.

jobs := make(chan Job, workers*2)

select {
case jobs <- job:
    // Задание принято.
case <-ctx.Done():
    return ctx.Err()
}
Не путайте очередь и хранилищеКанал живёт в памяти процесса. После перезапуска задания исчезнут. Если потеря недопустима, источник истины должен быть устойчивым: таблица в базе, журнал или внешний брокер.

Кто закрывает канал

Канал закрывает сторона, которая точно знает, что новых значений больше не будет, — обычно производитель. Воркеры только читают и завершаются после ok == false. Закрытие канала потребителем создаёт гонку с отправителем и рано или поздно приводит к панике.

Для конечной пачки схема простая: производитель отправляет все задания и закрывает канал; пул дочитывает остаток и возвращает результат. Для постоянно работающего сервиса канал может вообще не закрываться до остановки процесса — тогда жизненным циклом управляет контекст.

Какую ошибку считать фатальной

В примере первая ошибка останавливает весь пул. Это подходит для атомарной пачки, но не для независимых фоновых заданий. Там лучше возвращать результат по отдельному каналу, записывать статус каждого задания и продолжать работу после ожидаемых ошибок.

  • Ошибка входных данных относится к одному заданию.
  • Отмена сервиса останавливает всех воркеров.
  • Недоступность общей зависимости может остановить приём новых задач.
  • Паника в обработчике — дефект; её нельзя молча превращать в обычный результат.

Наблюдаемость без фреймворка

Для начала достаточно четырёх величин: длина очереди, число активных воркеров, длительность задания и количество исходов по типу. В логах полезны ID задания, попытка и причина завершения. Эти сигналы отвечают на главный вопрос: мы справляемся с входным потоком или только копим задержку?

Хорошая очередь не делает перегрузку невозможной. Она делает перегрузку видимой и управляемой.

Пул на канале — не замена Kafka или облачной очереди. Это маленький механизм для маленькой границы: один процесс, ограниченный параллелизм, допустимая потеря незавершённой работы. Пока граница именно такая, дополнительная магия только скрывает решения, которые всё равно придётся принять.