Приоритетные очереди в Go: рабочие приёмы для воркеров

Экспорт отчёта для одного пользователя ждёт в очереди, пока воркеры перемалывают ночную рассылку на сто тысяч писем. Пользователь смотрит на крутилку минуту, потом уходит. Задача была на две секунды, но она встала в общий канал за массовыми — и очередь без приоритетов сделала своё дело. Разберём, как в Go развести задачи по важности: от простого варианта на нескольких каналах до container/heap с динамическим приоритетом, и чем каждый вариант платит. Код проверялся на Go 1.22.

Речь только про приоритезацию внутри одного процесса: пул горутин-воркеров и источник задач. Персистентность, ретраи и распределённые очереди (Redis, RabbitMQ, NATS) — отдельная тема, здесь их не трогаем. Задача одна: когда воркер освобождается, он должен взять самую важную из ждущих, а не самую старую.

Подход 1: несколько каналов и select с приоритетом

Самое дешёвое решение, если уровней приоритета мало и они фиксированы. Делаете отдельный канал на каждый уровень и в воркере читаете их в порядке важности.

package main

import "fmt"

type Job struct {
	ID   int
	Kind string
}

func worker(high, low chan Job, done chan struct{}) {
	for {
		// Сначала пытаемся забрать срочную задачу без блокировки.
		select {
		case j := <-high:
			process(j)
			continue
		default:
		}
		// Срочных нет — берём любую, что придёт первой.
		select {
		case j := <-high:
			process(j)
		case j := <-low:
			process(j)
		case <-done:
			return
		}
	}
}

func process(j Job) {
	fmt.Printf("processing %s job %d\n", j.Kind, j.ID)
}

Первый select с default — это неблокирующая проверка: если в high что-то есть, берём и не доходим до второго select. Если пусто, ныряем во второй select, где ждём любую задачу или сигнал остановки.

Плата за простоту: приоритетов ровно столько, сколько каналов. Три уровня — три канала и вложенные проверки, которые быстро становятся нечитаемыми. Второй нюанс — голодание¹ низкого приоритета: пока в high непрерывно валятся задачи, low не обрабатывается вообще. Для отчётов против рассылки это может быть даже желаемым поведением, но осознавайте его.

Подход 2: container/heap с числовым приоритетом

Когда приоритет — это число (чем меньше, тем важнее) и уровней много, каналы не тянут. Нужна куча. В стандартной библиотеке есть container/heap — вы реализуете интерфейс heap.Interface, а пакет даёт логику вставки и извлечения за O(log n).

package main

import "container/heap"

type PriorityJob struct {
	ID       int
	Priority int // меньше — важнее
	index    int // позиция в куче, нужна heap.Fix
}

type JobQueue []*PriorityJob

func (q JobQueue) Len() int { return len(q) }
func (q JobQueue) Less(i, j int) bool {
	return q[i].Priority < q[j].Priority
}
func (q JobQueue) Swap(i, j int) {
	q[i], q[j] = q[j], q[i]
	q[i].index = i
	q[j].index = j
}

func (q *JobQueue) Push(x any) {
	job := x.(*PriorityJob)
	job.index = len(*q)
	*q = append(*q, job)
}

func (q *JobQueue) Pop() any {
	old := *q
	n := len(old)
	job := old[n-1]
	old[n-1] = nil // помогаем сборщику мусора
	*q = old[:n-1]
	return job
}

Дальше очередь оборачивается в структуру с мьютексом, потому что heap сам по себе не потокобезопасен. Воркеры и продюсеры дёргают её из разных горутин.

import (
	"container/heap"
	"sync"
)

type SafeQueue struct {
	mu    sync.Mutex
	queue JobQueue
	cond  *sync.Cond
}

func NewSafeQueue() *SafeQueue {
	q := &SafeQueue{}
	q.cond = sync.NewCond(&q.mu)
	return q
}

func (s *SafeQueue) Add(job *PriorityJob) {
	s.mu.Lock()
	heap.Push(&s.queue, job)
	s.mu.Unlock()
	s.cond.Signal() // будим один ждущий воркер
}

func (s *SafeQueue) Take() *PriorityJob {
	s.mu.Lock()
	defer s.mu.Unlock()
	for s.queue.Len() == 0 {
		s.cond.Wait() // отпускает мьютекс, пока ждём
	}
	return heap.Pop(&s.queue).(*PriorityJob)
}

Здесь sync.Cond заменяет канал: воркер в Take засыпает на Wait, если куча пуста, и просыпается по Signal от продюсера. Это классическая связка мьютекс плюс condition variable — если она вам непривычна, для остановки воркеров придётся добавить проверку флага после Wait, иначе горутины повиснут при завершении.

Плата: сложнее кода и ручная синхронизация. Зато приоритет — произвольное число, и можно менять его на лету через heap.Fix, если задача в очереди подорожала.

Куча решает, кого взять следующим. Но она ничего не знает про то, сколько задача уже ждёт. Без страховки от голодания приоритетная очередь превращается в машину, которая честно душит всё, что дешевле топ-приоритета.

Подход 3: heap плюс защита от голодания

Проблема чистой кучи: задача с низким приоритетом может ждать бесконечно, пока сыплются важные. Лечится старением — приоритет задачи со временем растёт. Вместо статичного числа считайте эффективный приоритет с поправкой на время ожидания.

import "time"

type AgingJob struct {
	ID       int
	Base     int       // исходный приоритет
	Enqueued time.Time // когда попала в очередь
	index    int
}

// Чем дольше ждём, тем меньше эффективное число (важнее).
func (j *AgingJob) effective(now time.Time) int {
	waited := now.Sub(j.Enqueued)
	// каждые 10 секунд ожидания снижают число на 1
	return j.Base - int(waited/(10*time.Second))
}

В Less сравниваете effective, а не Base. Тонкость: effective зависит от текущего времени, поэтому порядок в куче формально нарушается между вставками. На практике это терпимо, если сравнение вызывается только внутри операций heap с одним фиксированным now на операцию. Если нужна строгость — пересчитывайте приоритеты пакетно по таймеру и вызывайте heap.Init.

Подход 4: буферизованные каналы как приоритетный шлюз

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

Минус — диспетчер становится единой точкой, через которую проходит всё, и его пропускная способность ограничивает всю систему. При десятках тысяч задач в секунду это узкое место видно на профиле.

Когда что выбирать

Что ломается

• Забыли мьютекс вокруг heap. container/heap не потокобезопасен. Гонка проявится не сразу, а на нагрузке — паникой index out of range внутри Pop. Ловится флагом -race в тестах: запускайте go test -race.
• Cond.Wait без цикла. Проверять условие надо в for, а не в if — возможны ложные пробуждения и гонка между Signal и повторной проверкой. Замена for на if в Take даёт извлечение из пустой кучи и панику.
• Неограниченная очередь. Куча растёт, пока продюсеры быстрее воркеров. Память кончится тихо. Ставьте лимит на длину и решайте, что делать при переполнении: отбрасывать низкоприоритетное или блокировать продюсера.
• Приоритет решает всё, а латентность не мерится. Даже с идеальной кучей задача low может ждать минуты. Если у вас есть SLA на фоновые задачи, меряйте время от постановки до старта обработки по каждому уровню отдельно, иначе проблема голодания не видна в метриках.

Что настроить дальше

Начните с подхода 1, если уровней два-три и голодание low вам не мешает — это меньше всего кода и ошибок. Переходите на heap, когда уровней становится больше или приоритет вычисляется динамически. Старение добавляйте только когда в метриках увидите, что низкоприоритетные задачи реально висят дольше допустимого.

В проде смотрите на две вещи: длину очереди по уровням (растёт — воркеров мало или продюсер разогнался) и время ожидания в очереди по каждому приоритету (percentile p95 и p99, не среднее). Для тюнинга числа воркеров начните с runtime.NumCPU() для CPU-задач и заметно больше для IO-bound — точное число подбирается замером под вашу нагрузку.

¹ Голодание (starvation) — ситуация, когда задача бесконечно не получает ресурс, потому что его всё время забирают более приоритетные. Задача корректна, но никогда не выполняется.