Очереди задач в Go: создание, обработка и оптимизация

Онлайн-тренажер Python для начинающих

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

Начать курс

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

В современной разработке бэкенда часто возникают задачи, которые не требуют немедленного ответа пользователю: отправка email, генерация отчётов, обработка изображений, синхронизация с внешними API. Выполнять такие операции синхронно в рамках HTTP-запроса — плохая практика, так как это замедляет ответ и увеличивает нагрузку на сервер.

Очереди задач (job queues) решают эту проблему. В Go (Golang) встроенные механизмы конкурентности — горутины и каналы — позволяют строить надёжные и производительные очереди без внешних зависимостей. В этой статье мы разберём, как создать очередь задач с нуля, оптимизировать её с помощью worker pool и избежать типичных ошибок.

Основы: горутины и каналы

Go предоставляет два ключевых примитива для работы с очередями:

Горутины — лёгкие потоки, которые запускаются с помощью ключевого слова go.
Каналы — типизированные «трубы» для передачи данных между горутинами.

Простейшая очередь — это горутина, которая читает из канала и обрабатывает задачи:

package main

import (
    "fmt"
    "time"
)

func worker(jobs <-chan int) {
    for job := range jobs {
        fmt.Printf("Обработка задачи %d\n", job)
        time.Sleep(1 * time.Second) // имитация работы
    }
}

func main() {
    jobs := make(chan int, 10)

    // запускаем воркера
    go worker(jobs)

    // отправляем задачи
    for i := 1; i <= 5; i++ {
        jobs <- i
    }

    close(jobs) // закрываем канал после отправки всех задач
    time.Sleep(2 * time.Second)
}

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

Реализация Worker Pool в Go

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

Базовая реализация

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(500 * time.Millisecond)
    }
}

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

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

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

    // отправляем задачи
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs)

    wg.Wait() // ждём завершения всех воркеров
    fmt.Println("Все задачи выполнены")
}

Здесь мы используем sync.WaitGroup для ожидания завершения всех воркеров. Канал с буфером numJobs позволяет отправить все задачи без блокировки, пока воркеры не успели их забрать.

Динамическое добавление задач

В реальных приложениях задачи приходят асинхронно — например, по HTTP. Вот пример с обработчиком:

package main

import (
    "fmt"
    "net/http"
    "sync"
    "time"
)

type Task struct {
    ID   int
    Data string
}

var (
    taskQueue = make(chan Task, 100)
    wg        sync.WaitGroup
)

func worker(id int) {
    defer wg.Done()
    for task := range taskQueue {
        fmt.Printf("[Воркер %d] Обработка: %s (ID=%d)\n", id, task.Data, task.ID)
        time.Sleep(1 * time.Second)
    }
}

func handler(w http.ResponseWriter, r *http.Request) {
    // получаем задачу из запроса (упрощённо)
    task := Task{ID: time.Now().Nanosecond(), Data: "Новая задача"}
    taskQueue <- task
    w.Write([]byte("Задача поставлена в очередь"))
}

func main() {
    // запускаем 4 воркера
    for i := 1; i <= 4; i++ {
        wg.Add(1)
        go worker(i)
    }

    http.HandleFunc("/enqueue", handler)
    fmt.Println("Сервер запущен на :8080")
    http.ListenAndServe(":8080", nil)

    // при завершении сервера закрываем очередь
    close(taskQueue)
    wg.Wait()
}

Продвинутые техники: приоритеты, таймауты и graceful shutdown

Приоритетные очереди

Иногда задачи нужно обрабатывать в порядке приоритета, а не FIFO. В Go можно реализовать это через кучу (heap) или несколько каналов:

package main

import (
    "container/heap"
    "fmt"
)

type Job struct {
    priority int
    index    int
    action   func()
}

type PriorityQueue []*Job

func (pq PriorityQueue) Len() int { return len(pq) }

func (pq PriorityQueue) Less(i, j int) bool {
    return pq[i].priority > pq[j].priority // высокий приоритет — первым
}

func (pq PriorityQueue) Swap(i, j int) {
    pq[i], pq[j] = pq[j], pq[i]
    pq[i].index = i
    pq[j].index = j
}

func (pq *PriorityQueue) Push(x interface{}) {
    n := len(*pq)
    item := x.(*Job)
    item.index = n
    *pq = append(*pq, item)
}

func (pq *PriorityQueue) Pop() interface{} {
    old := *pq
    n := len(old)
    item := old[n-1]
    old[n-1] = nil
    item.index = -1
    *pq = old[0 : n-1]
    return item
}

func main() {
    pq := &PriorityQueue{}
    heap.Init(pq)

    heap.Push(pq, &Job{priority: 3, action: func() { fmt.Println("Низкий приоритет") }})
    heap.Push(pq, &Job{priority: 10, action: func() { fmt.Println("Высокий приоритет") }})

    for pq.Len() > 0 {
        job := heap.Pop(pq).(*Job)
        job.action()
    }
}