Приоритетные очереди задач в Go: сравниваем четыре подхода

Сценарий, который приходит почти в каждый бэкенд на Go: есть пул горутин, который разбирает задачи из канала. Сначала всё ровно — задачи короткие, воркеров хватает. Потом добавляется генерация PDF-отчёта на 30 секунд, и вот уже пользователь ждёт письмо-подтверждение две минуты, потому что перед ним в очереди встали три отчёта. Обычный chan Task тут не спасёт: канал в Go — это FIFO, он не умеет пропускать вперёд более важное.

Ниже — четыре способа сделать фоновую обработку с приоритетами: два на стандартной библиотеке (Go 1.22+), два на внешних брокерах. У каждого своя цена по коду, задержке и надёжности. В конце — таблица и критерий выбора, а не «зависит от задачи».

Что именно мы решаем

«Приоритет» в очереди задач — это гарантия, что при наличии готовой задачи высокого приоритета воркер возьмёт её раньше низкой, даже если низкая пришла первой. Разворачивать саму бизнес-логику задач мы не будем — фокус только на диспетчеризации по приоритетам. Ретраи, идемпотентность и хранение результата — отдельные темы, здесь их не трогаем.

Договоримся о трёх уровнях: high (пользователь ждёт ответа прямо сейчас), normal (обычная фоновая работа), low (отчёты, экспорт, чистка). Задача — структура с полем приоритета:

type Priority int

const (
    Low Priority = iota
    Normal
    High
)

type Task struct {
    Priority Priority
    Run      func() // саму работу упрощаем до функции
}

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

Самое простое, что работает без единой зависимости. Заводим по каналу на уровень и в воркере читаем через вложенный select с default: сначала пытаемся взять high, если пусто — normal, потом low.

type Dispatcher struct {
    high   chan Task
    normal chan Task
    low    chan Task
}

func (d *Dispatcher) worker() {
    for {
        // приоритетное чтение: high вне очереди
        select {
        case t := <-d.high:
            t.Run()
            continue
        default:
        }
        select {
        case t := <-d.high:
            t.Run()
        case t := <-d.normal:
            t.Run()
        case t := <-d.low:
            t.Run()
        }
    }
}

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

Цена. Приоритет здесь мягкий, а не строгий. Во втором select Go выбирает готовый case псевдослучайно — если high и low готовы одновременно, может уйти low. Строгую гарантию даёт только первый блок с default, и он покрывает лишь high. Для пяти уровней такая лестница из select превращается в нечитаемую конструкцию.

Подход 2. Куча под мьютексом (container/heap)

Когда уровней много или приоритет должен быть строгим, канал заменяют на кучу из стандартного пакета container/heap. Воркеры ждут на sync.Cond, пока в куче что-то появится.

import (
    "container/heap"
    "sync"
)

type taskHeap []Task

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.(Task)) }
func (h *taskHeap) Pop() any {
    old := *h
    n := len(old)
    t := old[n-1]
    *h = old[:n-1]
    return t
}

type PriorityQueue struct {
    mu   sync.Mutex
    cond *sync.Cond
    h    taskHeap
}

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

func (q *PriorityQueue) Add(t Task) {
    q.mu.Lock()
    heap.Push(&q.h, t)
    q.mu.Unlock()
    q.cond.Signal() // будим один спящий воркер
}

func (q *PriorityQueue) Take() Task {
    q.mu.Lock()
    for q.h.Len() == 0 {
        q.cond.Wait()
    }
    t := heap.Pop(&q.h).(Task)
    q.mu.Unlock()
    return t
}

Приоритет строгий: Pop всегда отдаёт самую важную задачу за O(log n). Задачи одного приоритета кучей не упорядочены по времени — при равных Less порядок не гарантирован. Если внутри уровня нужен FIFO, добавьте в структуру порядковый номер и сравнивайте его при равных приоритетах.

Куча даёт строгий приоритет, но вся очередь живёт в памяти одного процесса. Перезагрузили под — потеряли всё, что не успели обработать. Это цена, а не деталь.

Подход 3. Asynq поверх Redis

Если задачи нельзя терять при рестарте, in-memory очередь не подходит — нужен внешний брокер. Библиотека hibiken/asynq держит задачи в Redis и умеет приоритеты через именованные очереди с весами.

Вместо числового поля приоритета вы кладёте задачи в очереди с именами и задаёте им веса. Воркер разбирает очереди пропорционально весам, а флаг строгого приоритета заставляет сначала опустошить старшую очередь. Точные имена опций конфигурации сверяйте с README пакета — API между версиями менялся, и подставлять по памяти имя поля я не буду.

Цена. Появляется зависимость от Redis: ещё один компонент, который нужно мониторить и который сам может стать узким местом. Зато задачи переживают рестарт воркеров, есть встроенные ретраи, отложенный запуск и веб-панель мониторинга. Для очередей, где потеря задачи означает потерянный платёж или неотправленное письмо, это разумный обмен.

Подход 4. RabbitMQ с consumer priority

Когда очередь общая для нескольких сервисов на разных языках, а не только для Go, приоритеты логичнее держать в брокере. RabbitMQ поддерживает priority queues: очередь объявляется с максимальным уровнем приоритета, а сообщение публикуется с полем приоритета. Брокер сам отдаёт консьюмеру более важные сообщения раньше.

Цена. Самый тяжёлый вариант по инфраструктуре: отдельный кластер, настройка, эксплуатация. Приоритет в RabbitMQ применяется к сообщениям, которые ещё в очереди, — уже отданные консьюмеру и подтверждаемые обгонять нельзя. Точное поведение priority queues и ограничения (например, по числу уровней) описаны в официальной документции RabbitMQ — проверяйте под вашу версию брокера.

Сравнение

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

• Каналы + select — если задачи можно потерять при рестарте, а уровней два-три и «мягкий» приоритет вас устраивает. Внутрипроцессные фоновые дела: прогрев кеша, отложенная отправка метрик.
• container/heap — тот же in-memory сценарий, но уровней много или нужен строгий порядок. Например, планировщик, где важность вычисляется динамически.
• Asynq — задачи терять нельзя, инфраструктура уже на Redis, весь код на Go. Самый частый продовый выбор для одного сервиса.
• RabbitMQ — очередь общая для нескольких сервисов и языков, нужна маршрутизация сложнее, чем «важнее/менее важно».

Что ломается на проде

Первое — голодание низких приоритетов. Если high приходят непрерывно, low не выполнятся никогда. Лечится квотой: часть воркеров резервируют под low или после N задач high принудительно берут одну low.

Второе — рост очереди в памяти. У куч и каналов нет предела: если производитель быстрее потребителей, память течёт до OOM. Ставьте ограничение размера канала (буфер) или проверяйте длину кучи перед Add и отклоняйте лишнее.

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

Что делать дальше

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