Очереди задач в Rust: практическое руководство с примерами кода

Онлайн-тренажер Python для начинающих

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

Начать курс

Введение: зачем очереди задач в Rust?

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

В этой статье мы разберем три практических подхода к работе с очередями задач на Rust:

Использование стандартных каналов mpsc (multi-producer, single-consumer);
Многопоточные очереди с библиотекой crossbeam;
Асинхронные очереди на tokio.

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

1. Классические каналы: `std::sync::mpsc`

Стандартная библиотека Rust предоставляет модуль std::sync::mpsc — многопоточную очередь с одним потребителем. Это идеальный выбор, когда у вас есть несколько продюсеров (потоков, которые отправляют задачи) и один воркер (поток, который их выполняет).

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

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    // Создаем канал: sender (отправитель) и receiver (получатель)
    let (tx, rx) = mpsc::channel::<String>();

    // Запускаем поток-воркер
    let worker = thread::spawn(move || {
        for task in rx {
            println!("[Воркер] Выполняю задачу: {}", task);
            thread::sleep(Duration::from_millis(500));
            println!("[Воркер] Задача '{}' завершена!", task);
        }
    });

    // Создаем несколько продюсеров (в одном потоке для простоты)
    let producer = thread::spawn(move || {
        let tasks = vec![
            "Обработать файл A",
            "Отправить email",
            "Обновить базу данных",
            "Сгенерировать отчёт",
        ];
        for task in tasks {
            println!("[Продюсер] Отправляю: {}", task);
            tx.send(task.to_string()).unwrap();
            thread::sleep(Duration::from_millis(200));
        }
    });

    producer.join().unwrap();
    // После того, как sender (tx) будет уничтожен, receiver (rx) получит None и цикл завершится
    drop(tx); // закрываем канал со стороны продюсера
    worker.join().unwrap();
}

Особенности:

Канал mpsc буферизирован — задачи не блокируют отправителя, пока буфер не заполнен.
Только один получатель — это ограничение, но оно обеспечивает простоту и потокобезопасность.
При падении всех отправителей канал автоматически закрывается.

Несколько воркеров: ручное распределение

Если вам нужно несколько воркеров, mpsc не подходит напрямую. Решение — создать один канал и несколько потоков, которые будут конкурировать за получение задач. Однако это требует аккуратности: Receiver не реализует Sync, поэтому его нужно обернуть в Mutex или Arc.

use std::sync::{mpsc, Arc, Mutex};
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<String>();
    let rx = Arc::new(Mutex::new(rx));

    let mut workers = vec![];
    for id in 0..3 {
        let rx = Arc::clone(&rx);
        let worker = thread::spawn(move || {
            loop {
                let task = {
                    // Захватываем мьютекс и получаем задачу
                    let rx = rx.lock().unwrap();
                    rx.recv()
                };
                match task {
                    Ok(task) => {
                        println!("[Воркер {}] Выполняю: {}", id, task);
                        thread::sleep(Duration::from_millis(300));
                    }
                    Err(_) => {
                        println!("[Воркер {}] Канал закрыт, завершаюсь.", id);
                        break;
                    }
                }
            }
        });
        workers.push(worker);
    }

    // Отправляем задачи
    for i in 0..10 {
        tx.send(format!("Задача {}", i)).unwrap();
        thread::sleep(Duration::from_millis(100));
    }
    drop(tx); // закрываем канал

    for w in workers {
        w.join().unwrap();
    }
}

Минусы: каждая задача получается под мьютексом, что снижает производительность. Для высоконагруженных систем лучше использовать crossbeam.

2. Многопоточные очереди с `crossbeam`

Библиотека crossbeam предлагает более гибкие структуры, в том числе crossbeam_channel — многопоточную очередь с поддержкой множества потребителей (multi-consumer) без мьютексов. Она быстрее, чем std::sync::mpsc с оберткой, и предоставляет дополнительные возможности: выбор каналов (select!), таймауты и т.д.

Установка

# В Cargo.toml добавьте:
[dependencies]
crossbeam = "0.8"

Пример: пул воркеров с crossbeam

use crossbeam::channel;
use std::thread;
use std::time::Duration;

fn main() {
    // Создаем неограниченный канал (unbounded)
    let (tx, rx) = channel::unbounded::<String>();

    let mut workers = vec![];
    for id in 0..4 {
        let rx = rx.clone(); // клонируем получателя — это дешево
        workers.push(thread::spawn(move || {
            for task in rx {
                println!("[Воркер {}] Получил: {}", id, task);
                thread::sleep(Duration::from_millis(400 + id * 100));
                println!("[Воркер {}] Завершил: {}", id, task);
            }
        }));
    }

    // Продюсер
    for i in 0..8 {
        tx.send(format!("Задача #{}", i)).unwrap();
        thread::sleep(Duration::from_millis(150));
    }
    drop(tx); // закрываем канал

    for w in workers {
        w.join().unwrap();
    }
}

Преимущества crossbeam: