Очереди задач в Rust: полное руководство для начинающих

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

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

Rust предлагает несколько подходов к реализации очередей: от стандартных каналов из std::sync::mpsc до продвинутых асинхронных решений на базе tokio. Мы рассмотрим каждый из них и покажем, как выбрать правильный инструмент для вашей задачи.

Основы: очередь задач через mpsc каналы

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

Создание простейшей очереди

Рассмотрим базовый пример, где главный поток отправляет задачи, а отдельный поток их обрабатывает:

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

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

    // Запускаем поток-обработчик
    let handler = thread::spawn(move || {
        for task in rx {
            println!("Обрабатываю задачу: {}", task);
            thread::sleep(Duration::from_millis(500));
        }
    });

    // Отправляем задачи
    let tasks = vec!["Задание 1", "Задание 2", "Задание 3"];
    for task in tasks {
        tx.send(task.to_string()).unwrap();
        println!("Отправлено: {}", task);
    }

    // Закрываем канал, чтобы обработчик завершился
    drop(tx);
    handler.join().unwrap();
}

Этот код демонстрирует базовую механику: tx.send() помещает задачу в очередь, а получатель rx извлекает их в порядке поступления (FIFO). Обратите внимание, что канал автоматически блокирует производителя, если очередь переполнена, и блокирует потребителя, если очередь пуста.

Множественные производители

Одно из главных преимуществ mpsc — возможность клонировать отправителя для нескольких потоков:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel::<u32>();
    let mut handles = vec![];

    // Запускаем 3 потока-производителя
    for id in 0..3 {
        let tx_clone = tx.clone();
        handles.push(thread::spawn(move || {
            for i in 0..5 {
                let task = id * 100 + i;
                tx_clone.send(task).unwrap();
            }
        }));
    }

    // Закрываем оригинальный отправитель
    drop(tx);

    // Обрабатываем все задачи
    for received in rx {
        println!("Получена задача: {}", received);
    }

    for handle in handles {
        handle.join().unwrap();
    }
}

Каждый производитель отправляет свои задачи, а единый потребитель обрабатывает их в порядке поступления. Это классический паттерн "производитель-потребитель".

Продвинутая очередь с приоритетами

Стандартный канал mpsc не поддерживает приоритеты — задачи обрабатываются строго в порядке отправки. Однако мы можем реализовать очередь с приоритетами, используя BinaryHeap из стандартной библиотеки.

Реализация приоритетной очереди

use std::collections::BinaryHeap;
use std::cmp::Ordering;

#[derive(Debug, Eq, PartialEq)]
struct Task {
    priority: u8,
    data: String,
}

impl Ord for Task {
    fn cmp(&self, other: &Self) -> Ordering {
        // Чем меньше число, тем выше приоритет
        other.priority.cmp(&self.priority)
    }
}

impl PartialOrd for Task {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

fn main() {
    let mut queue = BinaryHeap::new();

    queue.push(Task { priority: 3, data: "Низкий приоритет".into() });
    queue.push(Task { priority: 1, data: "Высокий приоритет".into() });
    queue.push(Task { priority: 2, data: "Средний приоритет".into() });

    while let Some(task) = queue.pop() {
        println!("Обрабатываю: {} (приоритет: {})", task.data, task.priority);
    }
}

Мы реализуем трейты Ord и PartialOrd, чтобы BinaryHeap знал, как сортировать задачи. В результате задачи с наименьшим числовым приоритетом обрабатываются первыми.

Потокобезопасная приоритетная очередь

Для многопоточного использования обернем BinaryHeap в мьютекс:

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let queue = Arc::new(Mutex::new(BinaryHeap::new()));

    // Поток-производитель
    let producer_queue = queue.clone();
    let producer = thread::spawn(move || {
        let mut q = producer_queue.lock().unwrap();
        q.push(Task { priority: 1, data: "Срочная задача".into() });
        q.push(Task { priority: 5, data: "Несрочная задача".into() });
    });

    // Поток-потребитель
    let consumer_queue = queue.clone();
    let consumer = thread::spawn(move || {
        loop {
            let mut q = consumer_queue.lock().unwrap();
            if let Some(task) = q.pop() {
                println!("Обработка: {}", task.data);
            } else {
                break;
            }
        }
    });

    producer.join().unwrap();
    consumer.join().unwrap();
}

Обратите внимание: в реальных приложениях нужно добавить механизм уведомления потребителя о появлении новых задач (например, через условные переменные).

Асинхронные очереди с Tokio

Для высоконагруженных приложений (веб-серверы, обработчики событий) лучше использовать асинхронные очереди на базе библиотеки tokio. Они позволяют обрабатывать тысячи задач без создания отдельных потоков.

Установка зависимостей

Добавьте в Cargo.toml:

[dependencies]
tokio = { version = "1", features = ["full"] }

Асинхронный канал задач

use tokio::sync