Очереди задач в Go: практическое руководство

Введение: зачем нужны очереди задач в Go?

В современной разработке часто возникает необходимость обрабатывать задачи асинхронно, не блокируя основной поток приложения. Очереди задач (task queues) — это классический паттерн, который позволяет распределять работу между несколькими обработчиками (воркерами), управлять нагрузкой и обеспечивать отказоустойчивость.

Язык Go (Golang) предоставляет встроенные механизмы для реализации очередей задач: горутины (goroutines) и каналы (channels). В этой статье мы на практике разберем, как построить надежную очередь задач с помощью стандартной библиотеки Go, не прибегая к внешним брокерам сообщений (хотя в продакшене RabbitMQ или Redis могут быть полезны).

Основная часть

1. Базовый паттерн Worker Pool

Worker Pool — это набор фиксированного количества горутин (воркеров), которые слушают один канал и обрабатывают поступающие задачи. Это позволяет контролировать степень параллелизма и избежать перегрузки системы.

Рассмотрим простой пример: мы создаем пул из 3 воркеров, которые обрабатывают целые числа (симуляция работы).

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Воркер %d начал задачу %d\n", id, job)
        time.Sleep(time.Second) // симуляция работы
        fmt.Printf("Воркер %d завершил задачу %d\n", id, job)
    }
}

func main() {
    const numJobs = 5
    const numWorkers = 3

    jobs := make(chan int, numJobs)
    var wg sync.WaitGroup

    // Запускаем воркеров
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, &wg)
    }

    // Отправляем задачи
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs) // закрываем канал, чтобы воркеры завершились

    wg.Wait()
    fmt.Println("Все задачи выполнены")
}

Разбор кода:

• jobs — буферизированный канал, в который мы отправляем задачи.
• worker — функция, которая читает из канала и обрабатывает каждую задачу.
• sync.WaitGroup — ожидает завершения всех воркеров.
• После отправки всех задач канал закрывается (close(jobs)), что приводит к завершению цикла for range в воркерах.

Этот паттерн — основа для более сложных решений.

2. Очередь с приоритетами и таймаутами

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

package main

import (
    "context"
    "fmt"
    "time"
)

type Task struct {
    ID       int
    Priority int // чем меньше число, тем выше приоритет
    Data     string
}

func processTask(ctx context.Context, task Task) error {
    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-time.After(time.Duration(task.Priority) * 500 * time.Millisecond):
        fmt.Printf("Обработана задача %d: %s (приоритет %d)\n", task.ID, task.Data, task.Priority)
        return nil
    }
}

func main() {
    tasks := []Task{
        {ID: 1, Priority: 3, Data: "Низкий приоритет"},
        {ID: 2, Priority: 1, Data: "Высокий приоритет"},
        {ID: 3, Priority: 2, Data: "Средний приоритет"},
    }

    // Сортировка по приоритету (самые высокие — первыми)
    // В реальном проекте используйте сортировку или кучу (container/heap)
    for i := 0; i < len(tasks); i++ {
        for j := i + 1; j < len(tasks); j++ {
            if tasks[j].Priority < tasks[i].Priority {
                tasks[i], tasks[j] = tasks[j], tasks[i]
            }
        }
    }

    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    for _, task := range tasks {
        err := processTask(ctx, task)
        if err != nil {
            fmt.Printf("Задача %d отменена: %v\n", task.ID, err)
        }
    }
}

Важные моменты:

• Для полноценной очереди с приоритетами в Go используется container/heap (min-heap).
• context.WithTimeout позволяет задать общее время на обработку всех задач.
• Каждая задача может быть отменена, если не укладывается в лимит.

Этот подход полезен для систем, где важна скорость реакции на критические события.

3. Распределенная очередь с помощью каналов и Graceful Shutdown

В продакшене очередь должна корректно завершаться: дожидаться завершения текущих задач, не принимая новые. Реализуем graceful shutdown с использованием сигналов ОС.

package main

import (
    "fmt"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)

func main() {
    const numWorkers = 2
    jobs := make(chan int, 10)
    var wg sync.WaitGroup

    // Запуск воркеров
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobs {
                fmt.Printf("Воркер %d: обработка задачи %d\n", id, job)
                time.Sleep(2 * time.Second)
            }
        }(i)
    }

    // Канал для сигналов
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

    // Отправка задач в отдельной горутине
    go func() {
        for i := 1; i <= 10; i++ {
            select {
            case jobs <- i:
                fmt.Printf("Отправлена задача %d\n", i)
                time.Sleep(500 * time.Millisecond)
            case <-sigCh:
                fmt.Println("Получен сигнал остановки, прекращаем отправку")
                close(jobs)
                return
            }
        }
        close(jobs)
    }()

    // Ожидание сигнала в главной горутине
    <-sigCh
    fmt.Println("Завершаем