Очереди задач в Go: полное руководство для начинающих

Введение в очереди задач на Go

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

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

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

Прежде чем перейти к очередям, важно понять два фундаментальных понятия Go:

• Горутина (goroutine) — легковесный поток выполнения. Запускается ключевым словом go перед вызовом функции.
• Канал (channel) — типизированный «канал» для передачи данных между горутинами. Позволяет синхронизировать их работу.

Пример простейшей горутины и канала:

package main

import "fmt"

func main() {
    ch := make(chan string)

    go func() {
        ch <- "Привет из горутины!"
    }()

    msg := <-ch
    fmt.Println(msg)
}

В этом примере мы создали канал ch, запустили анонимную горутину, которая отправляет в него сообщение, и основная горутина получила это сообщение. Именно на этом механизме строятся очереди задач.

Реализация простой очереди задач

Самая простая очередь задач — это канал, в который мы помещаем задания, и одна или несколько горутин, которые их обрабатывают. Рассмотрим базовый пример:

package main

import (
    "fmt"
    "time"
)

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

func main() {
    tasks := make(chan int, 10) // буферизированный канал

    // запускаем одного воркера
    go worker(1, tasks)

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

    close(tasks) // закрываем канал
    time.Sleep(2 * time.Second) // даём время на обработку
}

Здесь worker — это функция, которая читает из канала tasks. Канал буферизирован (ёмкость 10), поэтому мы можем отправить несколько задач без блокировки. После закрытия канала цикл for range завершится автоматически.

Обратите внимание: мы использовали <-chan int — это канал только для чтения. Это хорошая практика, которая явно указывает, что воркер не должен отправлять данные в этот канал.

Буферизированные и небуферизированные каналы

В примере выше мы использовали буферизированный канал (make(chan int, 10)). Разница между типами каналов:

• Небуферизированный канал — отправка блокируется до тех пор, пока получатель не прочитает данные. Идеален для синхронизации.
• Буферизированный канал — отправка блокируется только когда буфер заполнен. Позволяет накапливать задачи и снижает задержки.

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

Пул воркеров (Worker Pool) на Go

Один воркер — хорошо, а несколько — лучше. Пул воркеров позволяет обрабатывать задачи параллельно, используя все ядра процессора. Вот как это реализовать:

package main

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

func worker(id int, tasks <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range tasks {
        fmt.Printf("Воркер %d обрабатывает задачу %d\n", id, task)
        time.Sleep(500 * time.Millisecond)
    }
}

func main() {
    const numWorkers = 3
    tasks := make(chan int, 10)

    var wg sync.WaitGroup

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

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

    close(tasks) // сигнал, что задач больше не будет
    wg.Wait()    // ждём завершения всех воркеров
    fmt.Println("Все задачи выполнены!")
}

Ключевые моменты:

• sync.WaitGroup — используется для ожидания завершения всех горутин. Каждый воркер вызывает wg.Done() при завершении.
• close(tasks) — закрытие канала сигнализирует воркерам, что новых задач не будет. Цикл for range завершается.
• defer wg.Done() — гарантирует, что счётчик уменьшится даже при панике.

Количество воркеров обычно выбирают равным runtime.NumCPU() или чуть больше, если задачи включают операции ввода-вывода.

Обработка ошибок и завершение с таймаутом

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

package main

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

type Task struct {
    ID   int
    Data string
}

type Result struct {
    TaskID int
    Err    error
}

func worker(id int, tasks <-chan Task, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range tasks {
        // имитация возможной ошибки
        if task.Data == "error" {
            results <- Result{TaskID: task.ID, Err: fmt.Errorf("ошибка в задаче %d", task.ID)}
            continue
        }
        time.Sleep(100 * time.Mill