Очереди и фоновые задачи на Go: горутины, каналы и продвинутые паттерны

Почему Go — идеальный язык для очередей и фоновых задач?

Если вы когда-нибудь писали бэкенд на Python или Node.js, то наверняка сталкивались с проблемой: приложение обрабатывает запрос, а в это время нужно отправить письмо, сгенерировать отчёт или обработать загруженный файл. Если делать это синхронно — пользователь ждёт. Если в отдельном потоке — начинаются танцы с блокировками и race conditions.

Go решает эту проблему элегантно и нативно. Горутины (goroutines) и каналы (channels) — это встроенные механизмы конкурентности, которые делают разработку очередей и фоновых задач на Go простой и безопасной. В этой статье мы разберём, как организовать фоновую обработку, используя как стандартную библиотеку, так и популярные инструменты вроде RabbitMQ. Вы увидите реальные примеры кода, которые можно сразу применить в своём проекте.

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

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

Простой пример: асинхронная отправка email

package main

import (
    "fmt"
    "time"
)

func sendEmail(email string) {
    time.Sleep(2 * time.Second) // имитация долгой операции
    fmt.Println("Email sent to", email)
}

func main() {
    fmt.Println("Start")
    go sendEmail("user@example.com") // фоновый запуск
    fmt.Println("Continue processing...")
    time.Sleep(3 * time.Second) // ждём завершения горутины
}

Здесь sendEmail выполняется в фоне, а основной поток не блокируется. Но что, если нужно обработать 1000 писем? Запускать 1000 горутин — не лучшая идея: они начнут конкурировать за ресурсы. Тут на помощь приходят каналы и паттерн Worker Pool.

Паттерн Worker Pool на Go

Worker Pool — это классический подход для организации очередей и фоновых задач на Go. У вас есть фиксированное количество воркеров (горутин), которые слушают один канал с задачами. Задачи поступают из основного потока или из HTTP-обработчика.

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

package main

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

type Task struct {
    ID   int
    Data string
}

func worker(id int, tasks <-chan Task, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range tasks {
        fmt.Printf("Worker %d processing task %d: %s\n", id, task.ID, task.Data)
        time.Sleep(1 * time.Second) // имитация работы
    }
}

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

    var wg sync.WaitGroup

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

    // Отправляем задачи
    for i := 1; i <= 10; i++ {
        tasks <- Task{ID: i, Data: fmt.Sprintf("Task %d", i)}
    }
    close(tasks) // закрываем канал, чтобы воркеры завершились

    wg.Wait()
    fmt.Println("All tasks completed")
}

Этот код создаёт 3 воркера, которые обрабатывают 10 задач. Буферизированный канал позволяет отправлять задачи без блокировки, пока буфер не заполнен. Паттерн идеально подходит для фоновых задач на Golang в веб-приложениях: вы принимаете запрос, кладёте задачу в канал и сразу отвечаете пользователю.

Очереди на Go с RabbitMQ

Для распределённых систем каналов в памяти недостаточно — нужно внешнее хранилище очередей. RabbitMQ — один из самых популярных брокеров сообщений. Рассмотрим, как подключить его к Go-приложению.

Установка и подключение

Сначала установите библиотеку amqp:

go get github.com/rabbitmq/amqp091-go

Producer (отправка задач)

package main

import (
    "log"
    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatal(err)
    }
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "task_queue", // имя очереди
        true,         // durable
        false,        // delete when unused
        false,        // exclusive
        false,        // no-wait
        nil,          // arguments
    )
    if err != nil {
        log.Fatal(err)
    }

    body := "Hello from Go!"
    err = ch.Publish(
        "",     // exchange
        q.Name, // routing key
        false,  // mandatory
        false,  // immediate
        amqp091.Publishing{
            DeliveryMode: amqp091.Persistent,
            ContentType:  "text/plain",
            Body:         []byte(body),
        })
    if err != nil {
        log.Fatal(err)
    }
    log.Printf(" [x] Sent %s", body)
}

Consumer (фоновый обработчик)

package main

import (
    "log"
    "time"
    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatal(err)
    }
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "task_queue",
        true,
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        log.Fatal(err)
    }

    msgs, err := ch.Consume(
        q.Name,
        "",
        false, // auto-ack
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        log.Fatal(err)
    }

    forever := make(chan struct{})

    go func() {
        for d := range msgs {
            log.Printf("Received a message: %s", d.Body)
            time.Sleep(1 * time.Second) // имитация обработки
            d.Ack(false)                // подтверждаем обработку
        }
    }()

    log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
    <-forever
}

Этот подход позволяет запускать несколько экземпляров consumer-приложений — RabbitMQ сам распределит задачи между ними. Отличное решение для обработки очередей на Go в микросервисной архитектуре.

Продвинутые техники: Graceful Shutdown и контексты

В реальных приложениях воркеры должны корректно завершаться при остановке сервера. Для этого используем context.Context:

package main

import (
    "context"
    "fmt"
    "os"
    "os/signal