Очереди и фоновые задачи на Rust: пишем асинхронного воркера за час

Очереди и фоновые задачи на Rust: пишем асинхронного воркера за час

Вы когда-нибудь задумывались, почему ваш веб-сервис иногда «тормозит», когда пользователь нажимает кнопку «Отправить»? Чаще всего виноваты тяжелые операции, которые выполняются прямо во время обработки запроса. Отправка email, генерация PDF, обработка изображений, интеграция с внешними API — всё это должно жить своей жизнью, а не блокировать ответ пользователю. Решение — очереди и фоновые задачи.

И если вы привыкли решать эту задачу на Python с Celery или RQ, то на Rust вас ждет сюрприз. Rust не просто быстр — он предоставляет уникальный контроль над памятью и потоками, позволяя писать сверхэффективные воркеры, которые потребляют минимум ресурсов. В этой статье мы разберем, как реализовать очереди и фоновые задачи на Rust, используя современные библиотеки и подходы. Вы узнаете, как построить надежную систему обработки задач, которая не упадет под нагрузкой.

Почему Rust для фоновых задач? Три неочевидных плюса

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

• Производительность без GC (сборщика мусора). Воркеры на Python могут «замирать» на паузах сборки мусора. Rust лишен этой проблемы — вы контролируете каждый байт памяти. Это критично для high-load систем, где важна каждая миллисекунда.
• Асинхронность "из коробки". Rust с tokio или async-std позволяет обрабатывать тысячи одновременных соединений на одном потоке. Вы можете запустить сотни воркеров, не боясь переполнения памяти.
• Надежность. Система владения (ownership) в Rust гарантирует, что вы не получите состояние гонки (data race) в фоновом воркере. Ошибки обрабатываются явно — никаких скрытых исключений.

Теперь, когда мы поняли «зачем», переходим к «как».

Выбираем брокер сообщений: RabbitMQ vs Redis vs NATS

Очередь задач — это не просто список. Это брокер сообщений, который хранит задачи и распределяет их между воркерами. В экосистеме Rust есть крейты под все популярные брокеры.

RabbitMQ + lapin: классика в мире очередей

RabbitMQ — это надежный монстр для enterprise-задач. Крейт lapin предоставляет чистый async API. Он идеален, если вам нужна маршрутизация сообщений, гарантии доставки и сложные топологии.

Redis + fred: простота и скорость

Redis — это швейцарский нож. С помощью fred (асинхронный клиент Redis) вы можете сделать очередь через списки (BLPOP) или Pub/Sub. Redis отлично подходит для простых сценариев: «отправил задачу — получил результат».

NATS + async-nats: легковесный брокер

NATS — это облачный брокер с минимальными задержками. Если вам нужно обрабатывать миллионы сообщений в секунду на маленьких инстансах, NATS ваш выбор.

В этой статье мы сосредоточимся на RabbitMQ, так как он даёт наиболее полный контроль. Но принципы будут те же для любого брокера.

Пишем очередь на Rust: пошаговый пример

Давайте создадим простую систему: генератор задач (producer) и воркер (consumer). Мы будем обрабатывать задачи по отправке email.

Шаг 1: Настройка проекта и зависимости

Создайте новый проект:

cargo new rust_queue_demo
cd rust_queue_demo

Добавьте зависимости в Cargo.toml:

[dependencies]
tokio = { version = "1", features = ["full"] }
lapin = "2.4"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tracing = "0.1"
tracing-subscriber = "0.3"
anyhow = "1"

Мы используем tokio как асинхронный рантайм, lapin для RabbitMQ, serde для сериализации задач и tracing для логирования.

Шаг 2: Определяем структуру задачи

Пусть наша задача — отправить email. Определим структуру:

use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EmailTask {
    pub to: String,
    pub subject: String,
    pub body: String,
}

impl EmailTask {
    pub fn new(to: &str, subject: &str, body: &str) -> Self {
        Self {
            to: to.to_string(),
            subject: subject.to_string(),
            body: body.to_string(),
        }
    }
}

Обратите внимание: мы используем Serialize и Deserialize. Это позволит нам превращать задачу в JSON для отправки в очередь.

Шаг 3: Создаём продюсер (отправитель задач)

Продюсер будет подключаться к RabbitMQ, объявлять очередь и отправлять сообщения.

use lapin::{options::*, types::FieldTable, BasicProperties, Connection, ConnectionProperties};
use anyhow::Result;

pub async fn publish_task(task: &EmailTask) -> Result<()> {
    // Подключаемся к RabbitMQ
    let addr = "amqp://guest:guest@localhost:5672";
    let conn = Connection::connect(addr, ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Объявляем очередь (создаст её, если не существует)
    let queue_name = "email_queue";
    channel
        .queue_declare(
            queue_name,
            QueueDeclareOptions::default(),
            FieldTable::default(),
        )
        .await?;

    // Превращаем задачу в JSON
    let payload = serde_json::to_vec(task)?;

    // Публикуем сообщение
    channel
        .basic_publish(
            "",          // exchange (пустой — используем default)
            queue_name,  // routing key
            BasicPublishOptions::default(),
            &payload,
            BasicProperties::default(),
        )
        .await?;

    println!("Задача отправлена: {:?}", task);
    Ok(())
}

Этот код — база. В реальном проекте вы бы добавили пул соединений, повторные попытки и мониторинг.

Шаг 4: Создаём воркер (потребитель задач)

Воркер будет слушать очередь и обрабатывать задачи. Здесь мы используем tokio для асинхронной обработки.

use futures::stream::StreamExt;
use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties};
use std::process;
use tracing::{info, error};

async fn process_email(task: &EmailTask) -> Result<()> {
    // Симулируем отправку email (задержка 2 секунды)
    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
    info!("Email отправлен на {}", task.to);