Очереди задач на Go: Обзор библиотек и инструментов

Очереди задач для Go (Golang): Обзор и сравнение инструментов

В современной веб-разработке и создании высоконагруженных сервисов на Go (Golang) практически всегда возникает необходимость выполнять задачи асинхронно, вне основного потока обработки HTTP-запросов. Отправка email-уведомлений, генерация PDF-отчетов, обработка изображений, взаимодействие с внешними API — все это требует выноса в фоновые процессы. Именно здесь на помощь приходят очереди задач (task queues).

Очередь задач — это механизм, который позволяет отложить выполнение операции, гарантировать её повторную обработку в случае сбоя и сбалансировать нагрузку между несколькими воркерами. В экосистеме Go существует несколько зрелых и активно развивающихся библиотек для решения этой задачи. В этой статье мы подробно разберем три самых популярных подхода: Asynq, Machinery и работу с Redis Streams напрямую.

1. Asynq: Простота и мощность на базе Redis

Asynq от разработчика hibiken — это, пожалуй, самая популярная библиотека для очередей задач в Go на данный момент. Она построена поверх Redis и предлагает богатый функционал «из коробки».

Ключевые особенности Asynq:

• Мониторинг: Встроенный веб-интерфейс (Asynqmon) для отслеживания очередей, выполнения задач и ошибок.
• Планирование: Возможность отложить выполнение задачи на определенное время (cron-like синтаксис).
• Приоритеты: Поддержка нескольких очередей с разными приоритетами.
• Retry и Dead Letters: Автоматические повторные попытки и очередь «мертвых» писем для невыполнимых задач.
• Middleware: Поддержка промежуточного ПО для логирования, метрик и т.д.

Пример использования Asynq:

package main

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

// Определяем тип задачи
const TypeEmailDelivery = "email:deliver"

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

    // Создаем задачу
    task, err := asynq.NewTask(TypeEmailDelivery, []byte(`{"user_id": 42, "template": "welcome"}`))
    if err != nil {
        log.Fatal(err)
    }

    // Отправляем задачу с задержкой в 5 минут
    info, err := client.Enqueue(task, asynq.ProcessIn(5*time.Minute))
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Задача [%s] поставлена в очередь", info.ID)
}

// Воркер для обработки задач
func handler(ctx context.Context, t *asynq.Task) error {
    log.Printf("Обработка задачи: %s", t.Type())
    // Здесь логика отправки email
    return nil
}

func startWorker() {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: "localhost:6379"},
        asynq.Config{Concurrency: 10},
    )

    mux := asynq.NewServeMux()
    mux.HandleFunc(TypeEmailDelivery, handler)

    if err := srv.Run(mux); err != nil {
        log.Fatal(err)
    }
}

Asynq идеально подходит для большинства проектов, где уже используется Redis. Он прост в настройке и предоставляет отличный UX для разработчика.

2. Machinery: Распределенные задачи с брокерами

Machinery — это более старая, но очень мощная библиотека, вдохновленная Python-библиотекой Celery. В отличие от Asynq, Machinery не привязан жестко к Redis и поддерживает несколько брокеров сообщений: Redis, RabbitMQ, AWS SQS и Memory.

Ключевые особенности Machinery:

• Гибкость брокеров: Легко переключаться между разными брокерами сообщений.
• Результаты задач: Встроенная поддержка сохранения и получения результатов выполнения задачи (например, в Redis).
• Цепочки задач: Можно создавать сложные пайплайны (цепочки, группы, хорды).
• Сохранение состояния: Поддержка хранения промежуточного состояния задачи.

Пример кода с Machinery:

package main

import (
    "context"
    "fmt"
    "github.com/RichardKnox/machinery/v1"
    "github.com/RichardKnox/machinery/v1/config"
    "github.com/RichardKnox/machinery/v1/tasks"
)

func Add(args ...int64) (int64, error) {
    sum := int64(0)
    for _, arg := range args {
        sum += arg
    }
    return sum, nil
}

func main