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

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

Проблема в том, что обычный буферизованный канал в Go — это строгая FIFO-очередь. Он не знает, что одна задача важнее другой. Разберём, как раздать задачам приоритеты, не наступив на классические грабли: голодание низкоприоритетных задач, гонки при чтении из нескольких каналов и потерю задач при завершении. Примеры на Go 1.22, всё из стандартной библиотеки плюс container/heap.

Сразу оговорка про угол: здесь мы решаем задачу приоритизации в рамках одного процесса. Распределённые очереди (Redis, RabbitMQ, NATS) с их приоритетами — отдельная тема, её не трогаем.

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

Самое очевидное решение — завести отдельный канал под каждый уровень приоритета и читать из них через select. Проблема в том, что select при нескольких готовых каналах выбирает случайный, а не первый по порядку. То есть наивный select приоритетов не даёт.

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

package main

import "context"

type Task struct {
	ID  int
	Run func()
}

type Dispatcher struct {
	high   chan Task // срочные: платежи, подтверждения
	normal chan Task // всё остальное
}

func NewDispatcher(bufSize int) *Dispatcher {
	return &Dispatcher{
		high:   make(chan Task, bufSize),
		normal: make(chan Task, bufSize),
	}
}

// worker читает задачи, отдавая предпочтение high
func (d *Dispatcher) worker(ctx context.Context) {
	for {
		// сначала пробуем срочные без блокировки
		select {
		case t := <-d.high:
			t.Run()
			continue
		default:
		}

		// high пуст — ждём любую задачу или отмену
		select {
		case t := <-d.high:
			t.Run()
		case t := <-d.normal:
			t.Run()
		case <-ctx.Done():
			return
		}
	}
}

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

Где это ломается

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

Подход 2: куча (heap) с одним каналом-сигналом

Когда приоритетов больше двух или нужен числовой приоритет (например, дедлайн), каналы не тянут. Здесь работает приоритетная очередь на базе container/heap: одна структура данных под мьютексом, из которой воркеры забирают задачу с максимальным приоритетом.

Канал отвечает за «есть работа / нет работы», а кто именно пойдёт следующим — решает куча. Не пытайтесь уместить приоритеты в сам канал: это его не та задача.

Реализация heap.Interface требует пяти методов. Пропустить хоть один — код не скомпилируется, так что подсказка компилятора здесь на вашей стороне.

package main

import (
	"container/heap"
	"sync"
)

type PriorityTask struct {
	Priority int // больше = важнее
	Run      func()
}

// taskHeap реализует heap.Interface
type taskHeap []PriorityTask

func (h taskHeap) Len() int           { return len(h) }
func (h taskHeap) Less(i, j int) bool { return h[i].Priority > h[j].Priority }
func (h taskHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }

func (h *taskHeap) Push(x any) {
	*h = append(*h, x.(PriorityTask))
}

func (h *taskHeap) Pop() any {
	old := *h
	n := len(old)
	item := old[n-1]
	*h = old[:n-1]
	return item
}

type Queue struct {
	mu     sync.Mutex
	heap   taskHeap
	signal chan struct{} // будит воркеров, когда появилась задача
}

func NewQueue() *Queue {
	return &Queue{signal: make(chan struct{}, 1)}
}

func (q *Queue) Enqueue(t PriorityTask) {
	q.mu.Lock()
	heap.Push(&q.heap, t)
	q.mu.Unlock()
	// неблокирующий сигнал: если он уже есть, второй не нужен
	select {
	case q.signal <- struct{}{}:
	default:
	}
}

// dequeue возвращает задачу и true, либо false если очередь пуста
func (q *Queue) dequeue() (PriorityTask, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.heap.Len() == 0 {
		return PriorityTask{}, false
	}
	return heap.Pop(&q.heap).(PriorityTask), true
}

Обратите внимание на Less: h[i].Priority > h[j].Priority — это max-heap, наверху самый важный. Если написать <, получите min-heap, и воркеры будут разбирать наименее важные задачи первыми. Это самая частая ошибка при первом заходе.

Цикл воркера с сигналом

func (q *Queue) worker(ctx context.Context) {
	for {
		t, ok := q.dequeue()
		if ok {
			t.Run()
			continue
		}
		// очередь пуста — ждём сигнала или отмены
		select {
		case <-q.signal:
		case <-ctx.Done():
			return
		}
	}
}

Ключевой момент: после получения сигнала воркер снова идёт в dequeue, а не читает задачу из канала. Сигнал буферизован размером 1 и служит только будильником. Если два воркера проснулись, а задача одна — второй получит false из dequeue и мирно уйдёт спать дальше. Гонки нет: доступ к куче под мьютексом.

Где это ломается

• Голодание при постоянном потоке важного. Как и в подходе с каналами, если high-задачи не кончаются, низкоприоритетные могут ждать очень долго. Лечится «старением»: периодически повышайте приоритет задач, лежащих дольше N секунд, пересчитывая кучу. Это усложняет код — вводите, только если голодание реально наблюдается.
• Мьютекс как точка контента. При десятках тысяч задач в секунду и большом пуле воркеров борьба за q.mu становится заметной. До этого масштаба одного мьютекса хватает.
• Задачи с равным приоритетом. Heap не стабилен — порядок задач с одинаковым Priority не гарантирован. Если нужен FIFO внутри уровня, добавьте в структуру монотонный счётчик и учитывайте его в Less при равенстве приоритетов.

Что выбрать

Правило простое: два-три фиксированных уровня без старения — берите каналы, там меньше кода и меньше шансов ошибиться в Less. Числовой приоритет, дедлайны, старение или изменение приоритета на лету — только heap.

Корректное завершение

Оба подхода завязаны на context для остановки. Но отмена контекста роняет воркеров немедленно — задачи, лежащие в очереди, теряются. Если это платежи, терять их нельзя.

Разделяйте два сигнала: ctx для жёсткой остановки и отдельный флаг «сток» (draining), при котором воркеры дорабатывают очередь до пустой, но новые задачи не принимаются. На уровне Enqueue проверяйте флаг и возвращайте ошибку, если очередь закрывается. Дожидайтесь воркеров через sync.WaitGroup перед выходом из процесса.

Что дальше

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