Очереди задач в Go: Обзор библиотек async, gocelery и machinery

Очереди задач для Go (Golang): Обзор и практика

В современных веб-приложениях и микросервисах часто возникает необходимость выполнять ресурсоёмкие или длительные операции асинхронно: отправка email, генерация отчётов, обработка изображений, взаимодействие с внешними API. Для этих целей используются очереди задач (task queues). В экосистеме Go (Golang) существует несколько зрелых библиотек, которые позволяют реализовать такую архитектуру. В этой статье мы рассмотрим три наиболее популярных инструмента: machinery, gocelery и asynq, разберём их особенности и приведём примеры кода.

Зачем нужна очередь задач в Go?

Очередь задач решает несколько ключевых проблем:

• Разделение ответственности: HTTP-обработчик не ждёт завершения фоновой задачи.
• Масштабирование: Вы можете запустить несколько воркеров (worker) на разных серверах.
• Надёжность: При сбое воркера задача не теряется и может быть повторно обработана.
• Планирование: Возможность отложить выполнение задачи на определённое время.

1. Machinery — мощная и гибкая библиотека

Machinery — это асинхронная очередь задач для Go, написанная с нуля. Она поддерживает несколько брокеров (Redis, RabbitMQ, AWS SQS) и хранилищ результатов (Redis, MongoDB, Memcache). Machinery предоставляет богатый API для создания задач, цепочек (chains), групп (groups) и повторяющихся задач.

Установка и пример

go get github.com/RichardKnox/machinery/v2

Пример простого воркера и клиента:

package main

import (
    "github.com/RichardKnox/machinery/v2"
    "github.com/RichardKnox/machinery/v2/config"
    "github.com/RichardKnox/machinery/v2/tasks"
)

func main() {
    // Конфигурация с Redis-брокером
    cnf := &config.Config{
        Broker:        "redis://localhost:6379",
        DefaultQueue:  "machinery_tasks",
        ResultBackend: "redis://localhost:6379",
    }

    server, err := machinery.NewServer(cnf)
    if err != nil {
        panic(err)
    }

    // Регистрация задачи
    err = server.RegisterTasks(map[string]interface{}{
        "add": func(a, b int) (int, error) {
            return a + b, nil
        },
    })
    if err != nil {
        panic(err)
    }

    // Запуск воркера
    worker := server.NewWorker("worker1", 2) // 2 concurrent tasks
    go worker.Launch()

    // Отправка задачи
    task := tasks.Signature{
        Name: "add",
        Args: []tasks.Arg{
            {Type: "int", Value: 5},
            {Type: "int", Value: 7},
        },
    }
    result, _ := server.SendTask(&task)
    // Ожидание результата
    if res, err := result.Get(10); err == nil {
        println("Result:", res.(int)) // Output: 12
    }
}

Плюсы: Поддержка сложных графов задач, возможность повторной попытки, мониторинг через веб-интерфейс.

Минусы: Относительно сложная настройка, документация иногда отстаёт от кода.

2. Asynq — современная очередь с богатым UI

Asynq — это библиотека от разработчиков Google, основанная на Redis. Она предоставляет удобный API, встроенное планирование, приоритеты задач и встроенный веб-интерфейс для мониторинга. Asynq активно развивается и имеет отличную документацию.

Установка и пример

go get github.com/hibiken/asynq

Пример простого клиента и воркера:

package main

import (
    "context"
    "log"
    "github.com/hibiken/asynq"
)

// Типизированная задача
const TypeEmailDelivery = "email:deliver"

func main() {
    // Клиент для отправки задач
    client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})
    defer client.Close()

    // Создание задачи
    task := asynq.NewTask(TypeEmailDelivery, []byte(`{"user_id": 42}`))
    info, err := client.Enqueue(task, asynq.MaxRetry(5))
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Task enqueued: %s", info.ID)

    // Воркер
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: "localhost:6379"},
        asynq.Config{Concurrency: 10},
    )

    // Обработчик задач
    mux := asynq.NewServeMux()
    mux.HandleFunc(TypeEmailDelivery, func(ctx context.Context, t *asynq.Task) error {
        log.Printf