Worker pool на Go: кто закрывает канал и как дождаться обработки

Задачи отправлены в канал, main завершился, и часть работы исчезла. Либо несколько работников закрывают один канал и получают panic. Для корректного завершения нужно разделить владельца входного канала и ожидание исполнителей.

Как разобрать ситуацию

В простом пуле производитель закрывает канал после последней отправки, а workers читают его через range. WaitGroup позволяет дождаться их окончания. Закрытие канала не означает, что вся обработка уже завершилась: сообщения могли быть прочитаны, но работа ещё продолжается. Это отдельное состояние жизненного цикла.

Пример и ожидаемый результат

Два работника суммируют задания 1, 2 и 3. Вход закрывается единственным производителем; общая сумма защищена mutex. После Wait программа должна вывести 6 независимо от порядка распределения задач между работниками.

package main

import (
	"fmt"
	"sync"
)

func main() {
	jobs := make(chan int)
	var wg sync.WaitGroup
	var mu sync.Mutex
	sum := 0
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for x := range jobs {
				mu.Lock()
				sum += x
				mu.Unlock()
			}
		}()
	}
	for _, x := range []int{1, 2, 3} {
		jobs <- x
	}
	close(jobs)
	wg.Wait()
	fmt.Println(sum)
}

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

Проверка перед применением

• Проверьте пустой набор и один элемент.
• Убедитесь, что только владелец отправки решает, когда новых задач больше не будет.
• Для отмены определите судьбу уже принятых, но не завершённых заданий: завершить, вернуть в очередь или пометить ошибкой.

Для рабочего пула добавьте метрики: сколько задач принято, выполняется, завершено и отклонено. Если вы видите только число отправленных сообщений, потеря при остановке останется невидимой. Небольшой тест завершения полезнее увеличения буфера канала: буфер меняет вместимость, но не создаёт гарантию доставки.

Границы решения

Канал находится в памяти процесса: при аварии его содержимое теряется. Для важных фоновых операций нужны внешнее хранилище, подтверждения и повторная обработка. Программа выполнена обычным go run; отдельный race detector этим запуском не включается.

Если базовый Go уже знаком и хочется связать HTTP, хранение данных и обработку отказов в одну систему, посмотрите наш платный курс

Микросервисы на Go: HTTP, PostgreSQL, gRPC, Kafka и Docker. Перед записью сравните программу со своей задачей и подготовкой.

Техническая справка: go.dev: документация, pkg.go.dev: документация.

Следующий шаг: Go backend: HTTP, конкурентность и устойчивость сервиса. По смежным вопросам: Как оборачивать ошибки в Go и не потерять машинную проверку причины, Плюс и амперсанд потерялись в URL Go-клиента: как собрать query.