🦀 Rust

Многопоточность и асинхронность в Rust: подробное руководство

Урок 32 / 35 • Полная теория для чтения

Что это такое и зачем? (аналогия из реальной жизни)

Представь, что ты — повар на кухне. Тебе нужно приготовить три блюда: суп, салат и десерт. У тебя есть только одна плита. Как ты будешь действовать?

Последовательный подход (без многопоточности): Ты ставишь суп вариться и стоишь рядом, ждешь, пока он закипит, потом варишь 20 минут, не отходя. Затем убираешь суп, ставишь чайник для десерта, ждешь... В итоге ты тратишь час, хотя большую часть времени просто стоял и смотрел на плиту. Это неэффективно.

Многопоточность (параллелизм): Ты нанимаешь двух помощников. Один варит суп, второй режет салат, третий занимается десертом. У каждого своя плита (свое ядро процессора). Все работают одновременно. Готово за 20 минут. Но есть проблема: если помощникам нужно пользоваться одной и той же разделочной доской (общий ресурс), они могут мешать друг другу.

Асинхронность (кооперативная многозадачность): Ты работаешь один, но очень умно. Ты поставил суп на плиту, включил таймер и, пока он варится, режешь салат. Как только таймер пикнул — ты переключился на суп, помешал его, поставил новый таймер и вернулся к салату. Ты не ждешь, а делаешь что-то полезное в ожидании. Это асинхронность: один поток (ты) переключается между задачами, не блокируясь на ожидании.

Важно: В Rust многопоточность и асинхронность — это разные инструменты. Многопоточность использует несколько ядер процессора для тяжелых вычислений. Асинхронность — для эффективного ожидания (сеть, диск, ввод пользователя) в одном потоке.

Теория: как это работает изнутри

Процессы и потоки

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

Rust позволяет создавать потоки через стандартную библиотеку std::thread. Каждый поток — это отдельный легковесный процесс, который может выполняться на своем ядре процессора.

Модель памяти и владение

Главная фишка Rust — система владения (ownership). Она работает и в многопоточности. Есть три правила:

Один владелец: У данных может быть только один владелец в каждый момент времени.
Заимствование: Можно дать данные "почитать" (&T) многим потокам одновременно, но только если данные не меняются.
Мутабельное заимствование: Если нужно менять данные, доступ должен быть исключительным (&mut T) — только один поток может писать.

Компилятор Rust проверяет эти правила на этапе компиляции. Если ты попытаешься отправить данные в другой поток, нарушив правила владения, программа просто не скомпилируется. Это гениально: ошибки многопоточности (гонки данных) отлавливаются до запуска программы!

Асинхронная модель

Асинхронность в Rust строится на Future (будущее). Future — это значение, которого еще нет, но которое появится позже. Представь, что ты заказал пиццу. Тебе дали номерок (Future). Ты не стоишь у стойки, а занимаешься своими делами. Когда пицца готова, тебя зовут (вызов колбэка).

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

Ключевое отличие от Python: В Python асинхронность встроена в язык (asyncio). В Rust асинхронность — это библиотека (tokio, async-std). Язык предоставляет только базовые примитивы (Future, async/await), а рантайм (цикл событий) выбирает программист.

Синтаксис в Rust: все формы и варианты

Создание потоков

Метод	Описание	Пример
`thread::spawn`	Создает новый поток и запускает в нем замыкание	`thread::spawn(\|\| { println!("Привет из потока!"); })`
`thread::Builder`	Позволяет настроить имя и размер стека потока	`Builder::new().name("мой-поток".into()).spawn(...)`
`JoinHandle`	Возвращается при spawn, позволяет дождаться завершения потока	`let handle = thread::spawn(...); handle.join().unwrap();`

Передача данных между потоками

Механизм	Описание	Когда использовать
`mpsc::channel`	Multiple Producer, Single Consumer — много пишут, один читает	Поток-рабочий отправляет результаты главному потоку
`Arc<Mutex<T>>`	Atomic Reference Counting + Mutex — разделяемый доступ с блокировкой	Несколько потоков читают и пишут в одни данные
`Arc<RwLock<T>>`	Блокировка чтения/записи — много читателей или один писатель	Данные часто читаются, редко пишутся
`AtomicBool, AtomicUsize`	Атомарные типы — без блокировок, для простых флагов и счетчиков	Флаг остановки потока, счетчик задач

Асинхронный синтаксис

Для асинхронности нужно добавить в проект библиотеку tokio (самая популярная).

// В Cargo.toml добавляем:
// [dependencies]
// tokio = { version = "1", features = ["full"] }

use tokio::time::{sleep, Duration};

// Асинхронная функция — она не выполняется сразу, а возвращает Future
async fn готовить_кофе() -> String {
    println!("Начинаю варить кофе...");
    // await — точка переключения: поток не блокируется, а ждет
    sleep(Duration::from_secs(2)).await;
    println!("Кофе готов!");
    "☕ Горячий кофе".to_string()
}

#[tokio::main]  // Этот макрос создает асинхронный рантайм
async fn main() {
    // Вызов async функции — это просто создание Future
    let future = готовить_кофе();
    
    // Чтобы Future выполнился, нужно его await'ить
    let результат = future.await;
    println!("Получено: {}", результат);
}

Примеры кода — от простого к сложному

Пример 1: Создание и ожидание потока

use std::thread;
use std::time::Duration;

fn main() {
    println!("Главный поток начал работу");
    
    // Создаем новый поток
    // move — забирает переменные из внешней среды в поток
    let handle = thread::spawn(move || {
        for i in 1..5 {
            println!("Поток-помощник: считаю до {}", i);
            // Засыпаем на 1 секунду, чтобы имитировать работу
            thread::sleep(Duration::from_millis(500));
        }
    });
    
    // Главный поток тоже работает
    for i in 1..3 {
        println!("Главный поток: делаю свои дела {}", i);
        thread::sleep(Duration::from_millis(300));
    }
    
    // Ждем завершения потока-помощника
    // join() возвращает Result, unwrap() игнорирует ошибки
    handle.join().unwrap();
    
    println!("Все потоки завершены!");
}

Пример 2: Каналы для передачи данных (mpsc)

use std::thread;
use std::sync::mpsc;  // multiple producer, single consumer

fn main() {
    // Создаем канал: tx — передатчик, rx — приемник
    let (tx, rx) = mpsc::channel();
    
    // Поток-отправитель
    thread::spawn(move || {
        let сообщения = vec!["Привет", "Как дела?", "Пока!"];
        for msg in сообщения {
            println!("Отправляю: {}", msg);
            // send() отправляет сообщение в канал
            tx.send(msg).unwrap();
            thread::sleep(std::time::Duration::from_secs(1));
        }
    });
    
    // Главный поток — получатель
    // recv() ждет, пока придет сообщение (блокирует поток)
    for получено in rx {
        println!("Получено: {}", получено);
    }
    
    // Когда передатчик закрыт, цикл завершается
    println!("Канал закрыт, все сообщения получены");
}

Пример 3: Разделяемый доступ через Arc и Mutex

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

fn main() {
    // Создаем счетчик, обернутый в Mutex и Arc
    // Arc — атомарный счетчик ссылок, позволяет разделить данные между потоками
    // Mutex — мьютекс, гарантирует, что только один поток имеет доступ к данным
    let счетчик = Arc::new(Mutex::new(0));
    
    let mut handles = vec![];
    
    for _ in 0..10 {
        // Клонируем Arc — увеличивается счетчик ссылок
        let счетчик_копия = Arc::clone(&счетчик);
        
        let handle = thread::spawn(move || {
            // lock() захватывает мьютекс — если занято, поток ждет
            let mut значение = счетчик_копия.lock().unwrap();
            *значение += 1;  // Увеличиваем счетчик
            // При выходе из области видимости lock() автоматически освобождается
        });
        
        handles.push(handle);
    }
    
    // Ждем все потоки
    for handle in handles {
        handle.join().unwrap();
    }
    
    // Читаем финальное значение
    println!("Итоговый счетчик: {}", *счетчик.lock().unwrap());
    // Должно быть 10 — каждый поток увеличил на 1
}

Пример 4: Асинхронная загрузка данных (с tokio)

use tokio::time::{sleep, Duration};

// Асинхронная функция, имитирующая загрузку из интернета
async fn загрузить_страницу(номер: u32) -> String {
    println!("Начинаю загрузку страницы {}", номер);
    // Имитируем задержку сети
    sleep(Duration::from_millis(500 * номер as u64)).await;
    println!("Страница {} загружена!", номер);
    format!("Содержимое страницы {}", номер)
}

#[tokio::main]
async fn main() {
    println!("Начинаем асинхронную загрузку...");
    
    // Создаем несколько Future, но не выполняем их
    let задача1 = загрузить_страницу(1);
    let задача2 = загрузить_страницу(2);
    let задача3 = загрузить_страницу(3);
    
    // Выполняем все задачи параллельно через join!
    // Все три загрузки будут выполняться одновременно
    let (результат1, результат2, результат3) = tokio::join!(задача1, задача2, задача3);
    
    println!("Результаты:");
    println!("1: {}", результат1);
    println!("2: {}", результат2);
    println!("3: {}", результат3);
    
    // Общее время — около 1.5 секунд (максимальная задержка), а не 3 секунды!
}

Пример 5: Асинхронный TCP-сервер

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> Result<(), Box> {
    // Слушаем порт 8080
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Сервер запущен на порту 8080");
    
    // Принимаем соединения в цикле
    loop {
        // accept() — асинхронный, не блокирует сервер
        let (mut socket, адрес) = listener.accept().await?;
        println!("Новое соединение от: {}", адрес);
        
        // Для каждого соединения запускаем асинхронную задачу
        tokio::spawn(async move {
            let mut buf = [0; 1024];
            
            // Читаем данные от клиента
            match socket.read(&mut buf).await {
                Ok(n) if n > 0 => {
                    let получено = String::from_utf8_lossy(&buf[..n]);
                    println!("Получено: {}", получено);
                    
                    // Отправляем ответ
                    let ответ = "Привет от Rust сервера!
";
                    socket.write_all(ответ.as_bytes()).await.unwrap();
                }
                _ => println!("Ошибка чтения или соединение закрыто"),
            }
        });
    }
}

Пример 6: Select! — выбор из нескольких Future

use tokio::time::{sleep, Duration};

async fn задача_долгая() -> &'static str {
    sleep(Duration::from_secs(5)).await;
    "Долгая задача завершена"
}

async fn задача_быстрая() -> &'static str {
    sleep(Duration::from_secs(2)).await;
    "Быстрая задача завершена"
}

#[tokio::main]
async fn main() {
    println!("Запускаем гонку задач...");
    
    // select! ждет первую завершившуюся задачу
    // Остальные задачи отменяются (drop)
    let результат = tokio::select! {
        результат = задача_долгая() => {
            println!("Выиграла долгая задача");
            результат
        }
        результат = задача_быстрая() => {
            println!("Выиграла быстрая задача");
            результат
        }
    };
    
    println!("Результат: {}", результат);
    // Выведет: "Выиграла быстрая задача"
}

Пример 7: Пул потоков для параллельных вычислений

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

fn main() {
    let start = Instant::now();
    
    // Создаем пул из 4 потоков
    let num_threads = 4;
    let (tx, rx) = mpsc::channel();
    
    // Разбиваем работу на части
    let tasks: Vec = (1..=100).collect();
    let chunk_size = tasks.len() / num_threads;
    
    for i in 0..num_threads {
        let tx = tx.clone();
        let start_idx = i * chunk_size;
        let end_idx = if i == num_threads - 1 {
            tasks.len()
        } else {
            (i + 1) * chunk_size
        };
        
        let chunk = tasks[start_idx..end_idx].to_vec();
        
        thread::spawn(move || {
            let sum: u64 = chunk.iter().sum();
            tx.send(sum).unwrap();
        });
    }
    
    // Собираем результаты
    let mut total = 0;
    for _ in 0..num_threads {
        total += rx.recv().unwrap();
    }
    
    let duration = start.elapsed();
    println!("Сумма чисел от 1 до 100: {}", total);
    println!("Время выполнения: {:?}", duration);
    // Должно быть 5050
}

Сравнение с Python: чем похоже и чем отличается

Характеристика	Rust	Python
Потоки (threading)	Настоящие системные потоки. Работают параллельно на многоядерных процессорах. Безопасность на уровне компиляции.	Потоки из-за GIL (Global Interpreter Lock) не выполняются параллельно для CPU-задач. Подходят только для I/O.
Асинхронность	Библиотечная (tokio, async-std). Нет встроенного цикла событий. Максимальная производительность.	Встроенная (asyncio). Проще для новичков, но медленнее.
Безопасность	Компилятор проверяет все на этапе компиляции. Гонки данных невозможны (если не использовать unsafe).	Ошибки многопоточности (race conditions) проявляются только во время выполнения. Нужны блокировки вручную.
Синтаксис async/await	`async fn foo() -> T` + `.await`	`async def foo() -> T` + `await`
Производительность	Очень высокая. Нулевая стоимость абстракций. Асинхронность сравнима с C++.	Низкая для CPU-задач. Асинхронность быстрее потоков, но медленнее Rust.
Сложность	Высокая. Нужно понимать владение, lifetimes, Send/Sync traits.	Низкая. Проще начать, сложнее отлаживать ошибки.

Совет: Если ты знаешь Python и хочешь изучить Rust, начни с асинхронности — она концептуально похожа. Многопоточность в Rust сложнее из-за системы владения, но зато безопаснее.

Типичные ошибки и подводные камни

1. Забыл move в замыкании

let data = vec![1, 2, 3];
// Ошибка: data может жить меньше, чем поток
thread::spawn(|| {
    println!("{:?}", data);
});
// Правильно:
thread::spawn(move || {
    println!("{:?}", data);
});

2. Взаимная блокировка (deadlock)

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

let data1 = Arc::new(Mutex::new(0));
let data2 = Arc::new(Mutex::new(0));

// Поток 1 захватывает data1, потом data2
// Поток 2 захватывает data2, потом data1
// Если они выполняются одновременно — deadlock!

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

3. Забыл обработать PoisonError

let mutex = Mutex::new(0);
let guard = mutex.lock().unwrap();  // Если другой поток запаниковал, unwrap вызовет панику
// Лучше использовать if let или match
if let Ok(mut guard) = mutex.lock() {
    *guard += 1;
}

4. Блокировка в асинхронном коде

// ПЛОХО: блокирующий sleep в асинхронной функции
async fn плохо() {
    std::thread::sleep(Duration::from_secs(1));  // Блокирует ВЕСЬ поток!
}

// ХОРОШО: асинхронный sleep
async fn хорошо() {
    tokio::time::sleep(Duration::from_secs(1)).await;  // Не блокирует
}

5. Слишком много потоков

Создание потока — дорогая операция (около 2 МБ на стек). Для тысяч задач используй асинхронность или пул потоков.

Лучшие практики и советы профессионалов

Используй асинхронность для I/O, потоки — для CPU. Если программа ждет сеть или диск — async. Если считает факториалы — потоки.
Не создавай потоки вручную для простых задач. Используй rayon для параллельных итераторов или tokio::spawn для асинхронных задач.
Минимизируй захват Mutex. Держи блокировку как можно меньше времени. Не делай внутри блокировки I/O или сложных вычислений.
Используй Arc<AtomicBool> для флагов остановки. Это быстрее и безопаснее, чем Mutex для простых значений.
Тестируй многопоточный код с loom. Это библиотека для проверки модели памяти Rust, помогает найти редкие баги.
Для асинхронности используй tokio. Это стандарт де-факто. async-std тоже хорош, но сообщество выбрало tokio.
Не используй unsafe для многопоточности. Ты не сможешь гарантировать безопасность. Доверься компилятору.
Измеряй производительность. Многопоточность не всегда быстрее. Иногда накладные расходы на синхронизацию перевешивают выгоду.

Вопросы для самопроверки

Вопрос: В чем разница между многопоточностью и асинхронностью?

Ответ: Многопоточность использует несколько ядер процессора для параллельного выполнения. Асинхронность — это эффективное переключение между задачами в одном потоке, когда задачи ждут (сеть, диск).
Вопрос: Что такое Send и Sync в Rust?

Ответ: Send — типаж, означающий, что значение можно передать в другой поток. Sync — типаж, означающий, что на значение можно ссылаться из нескольких потоков (&T: Send). Компилятор автоматически реализует их для безопасных типов.
Вопрос: Почему в Rust нет гонок данных (data races) в безопасном коде?

Ответ: Система владения гарантирует, что в каждый момент времени у данных либо один мутабельный владелец, либо много иммутабельных. Компилятор проверяет это на этапе компиляции.
Вопрос: Что делает Arc::clone()?

Ответ: Увеличивает счетчик ссылок на данные. Не копирует сами данные, только указатель. Когда все Arc уничтожены, данные освобождаются.
Вопрос: В чем разница между Mutex и RwLock?

Ответ: Mutex разрешает только одному потоку иметь доступ (чтение или запись). RwLock разрешает много читателей ИЛИ одного писателя. RwLock эффективнее, если чтений гораздо больше, чем записей.
Вопрос: Что такое #[tokio::main]?

Ответ: Это макрос, который превращает обычную функцию main в асинхронную. Он создает tokio-рантайм (цикл событий) и запускает в нем асинхронную функцию.
Вопрос: Что произойдет, если в асинхронной функции вызвать std::thread::sleep()?

Ответ: Заблокируется весь поток, в котором работает асинхронный рантайм. Другие задачи не смогут выполняться. Нужно использовать tokio::time::sleep().
Вопрос: Как отменить асинхронную задачу?

Ответ: Future отменяется, когда его дропают (drop). Например, если использовать tokio::select! и одна из веток выиграла, остальные Future дропаются. Можно также использовать tokio_util::CancellationToken.
Вопрос: Что такое mpsc и чем отличается от broadcast?

Ответ: mpsc (multi-producer, single-consumer) — один получатель. broadcast — много получателей, каждое сообщение получают все подписчики.
Вопрос: Почему в Python есть GIL, а в Rust нет?

Ответ: GIL в Python упрощает управление памятью (сборщик мусора). Rust использует систему владения без сборщика мусора, поэтому GIL не нужен. Rust может полноценно использовать многоядерные процессоры.

Итог и что изучить дальше

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

Как создавать потоки и обмениваться данными между ними
Как писать асинхронный код с tokio
Как избегать типичных ошибок
Чем Rust отличается от Python в этой области

Rust дает уникальное преимущество: безопасность многопоточности на уровне компиляции. Это значит, что многие ошибки, которые в других языках проявляются только в продакшене, в Rust отлавливаются до запуска программы.

Что изучить дальше:

Rayon — библиотека для параллельных итераторов. Позволяет легко распараллелить обработку коллекций.
Tokio подробнее — изучи TcpStream, UdpSocket, файловый I/O, таймеры.
Асинхронные стримы — Stream trait, аналогичный итераторам, но асинхронный.
Pin и Unpin — продвинутая тема, необходимая для понимания работы Future.
Бенчмаркинг — измеряй производительность своего кода с criterion.
Акторы — паттерн проектирования для многопоточности (библиотека actix).

Помни: лучший способ освоить многопоточность — писать код. Начни с простого TCP-чата, потом добавь пул потоков для обработки сообщений, затем перепиши на асинхронность. Сравни производительность и сложность. Удачи!

← Предыдущая Паттерны проектирования в Rust: подробное руководс Вперёд → Работа с сетью и HTTP API в Rust: подробное руково