Потоки и синхронные каналы в Rust

ОБЯЗАТЕЛЬНОДЛЯ НОВИЧКОВ

Разработчику Архитектору

Потоки и синхронные каналы в Rust

О чём эта глава

В бэкенде и системных утилитах почти сразу встаёт вопрос: как запускать несколько задач одновременно и при этом не ловить хаос в данных. Здесь и появляются потоки, каналы и примитивы синхронизации.

Поток ОС (std::thread) — отдельная "дорожка" выполнения с собственным стеком. Потоки могут работать параллельно на разных ядрах процессора. Это отличается от async (отдельная глава): там много задач по очереди отдают управление runtime, пока одна ждёт диск или сеть.

В Rust оба подхода используют регулярно. Компилятор через трейты Send и Sync запрещает передавать между потоками типы, для которых это небезопасно — так в safe-коде отсекаются типичные data race (гонки данных).

База владения: типы и владение. Общие примитивы: важные трейты.

Поток и задача — короткая аналогия

Понятие	Аналогия
Процесс	Отдельная программа с памятью
Поток	Рабочий внутри процесса; потоки делят память процесса
async-задача	Запись в планировщике: "когда сокет готов — продолжить эту функцию"

Один HTTP-сервер на Tokio может обслуживать тысячи соединений на небольшом числе потоков. Тяжёлый расчёт на всех ядрах CPU лучше разнести по std::thread или библиотеке rayon.

Когда потоки, когда async

Сценарий	Подход
Много ожидания сети или диска в одном сервисе	`async` + Tokio
CPU-bound расчёты на всех ядрах	`std::thread` или rayon
Фоновая работа из обычного sync-кода (CLI, часть GUI)	`thread::spawn`
Блокирующая библиотека внутри async-сервера	`tokio::task::spawn_blocking`

Async снижает стоимость ожидания I/O — пока один запрос ждёт ответа сети, runtime переключается на другой. Async не ускоряет чистую математику на CPU: для этого нужны потоки или пул rayon.

Запуск потока и `join`

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

fn main() {
    let handle = thread::spawn(|| {
        thread::sleep(Duration::from_millis(100));
        42
    });

    let answer = handle.join().expect("поток завершился с паникой");
    println!("{answer}");
}

Разбор:

thread::spawn(closure) — создаёт поток и сразу начинает выполнять замыкание.
Замыкание || { ... } возвращает 42 — это значение попадёт в join.
handle.join() — главный поток ждёт завершения дочернего. Тип: Result<T, Box<dyn Any + Send>>. Если внутри потока была panic, join вернёт Err.
thread::sleep — блокирует только этот поток ОС на 100 мс.

Передача данных в поток — `move`

По умолчанию замыкание заимствует переменные из окружения. Для передачи владения в другой поток нужно move:

let data = vec![1, 2, 3];
thread::spawn(move || {
    println!("{:?}", data);
});
// здесь `data` уже нельзя использовать — владение ушло в поток

Разбор:

vec![1, 2, 3] создаёт вектор во владении текущего потока.
Ключевое слово move переносит владение data в замыкание нового потока.
Без move замыкание попыталось бы заимствовать data, что небезопасно при выходе из области видимости.
После spawn исходная переменная data недоступна в родительском потоке - это защита от гонок владения.
println!("{:?}", data) использует формат Debug для печати содержимого вектора.

Компилятор проверит, что data реализует Send. Если бы это был Rc (счётчик ссылок без атомарности), spawn выдал бы ошибку: Rc нельзя безопасно делить между потоками.

Канал mpsc

mpsc = multi-producer, single-consumer — много отправителей, один получатель. Стандартный модуль: std::sync::mpsc.

Идея: потоки передают владение сообщением по очереди (как конвейер) — у получателя одна копия данных, без совместной записи в общую структуру.

Код ITЗагрузка примера кода…

Имя	Роль
`tx`	`Sender` — конец "отправки"
`rx`	`Receiver` — конец "приёма"
`send(value)`	Перемещает `value` в очередь; блокируется, если буфер переполнен (у `channel()` буфер неограничен)
`recv()`	Блокирующее чтение; ждёт сообщение

Дополнительные возможности:

tx.clone() — второй отправитель в другом потоке.
recv_timeout(dur) — ждать не дольше заданного времени.
try_recv() — проверить очередь без долгого ожидания (удобно в цикле событий).
rx.iter() — итератор: каждый элемент — следующее сообщение, пока все Sender не уничтожены.

Сообщениями передают String, Vec, структуры — всё, что владеет данными целиком. Сырые указатели без договорённостей между потоками в учебном safe-коде не используют.

Несколько отправителей через tx.clone():

let (tx, rx) = mpsc::channel();
let tx2 = tx.clone();

thread::spawn(move || tx.send(1).unwrap());
thread::spawn(move || tx2.send(2).unwrap());
drop(tx);
drop(tx2);

let mut values = vec![];
while let Ok(v) = rx.recv() {
    values.push(v);
}

Разбор:

tx.clone() создаёт второй Sender, указывающий на ту же очередь.
Каждый поток отправляет своё сообщение независимо.
drop(tx) и drop(tx2) закрывают отправителей, когда работа завершена.
Цикл while let Ok(v) = rx.recv() читает все сообщения до закрытия канала.
Так реализуют схему multi-producer -> single-consumer.

Общее изменяемое состояние — `Arc` + `Mutex`

Каналы хороши, когда владение переезжает от потока к потоку. Если нужен общий счётчик или кэш, применяют разделяемое состояние:

Код ITЗагрузка примера кода…

Тип	Зачем
`Mutex<T>`	В каждый момент только один поток держит `&mut T` (эксклюзивный доступ).
`Arc<Mutex<T>>`	`Arc` — атомарный счётчик ссылок, можно клонировать в несколько `spawn`; `Mutex` — защита данных.
`lock().unwrap()`	Захват; при панике в другом потоке с удержанным lock вернётся `PoisonError` (редко).
*`n += 1`**	Разыменование `MutexGuard` как изменяемой ссылки на число.

Итог печатает 4: четыре потока по разу увеличили счётчик.

RwLock — много читателей или один писатель (удобно для кэша). AtomicUsize — счётчик без блокировки для простых операций += 1.

Счётчик на атомиках без Mutex:

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

let counter = Arc::new(AtomicUsize::new(0));
let c = Arc::clone(&counter);

thread::spawn(move || {
    c.fetch_add(1, Ordering::Relaxed);
});

Разбор:

AtomicUsize хранит число, которое можно безопасно менять из нескольких потоков.
fetch_add(1, Ordering::Relaxed) атомарно увеличивает значение на 1.
Arc позволяет разделить один атомарный счётчик между потоками.
Для простых метрик это быстрее, чем Mutex на каждое += 1.
Ordering выбирают по требованиям к видимости изменений между потоками.

Send и Sync — что проверяет компилятор

Трейт	Смысл
`Send`	значение можно переместить в другой поток
`Sync`	на `&T` можно смотреть из нескольких потоков одновременно (ссылка `&T` сама `Send`)

Примеры:

i32, String, Vec<T> (если T: Send) — Send.
Rc<T> — не Send: счётчик ссылок без атомарных операций.
Arc<Mutex<T>> — Send, если T: Send.

Ошибка компилятора вида Rc<dyn Fn()> cannot be sent between threads — защита: вы пытались разделить тип, не предназначенный для многопоточности.

Потоки внутри async (Tokio)

В async fn нельзя долго блокировать worker runtime — остальные запросы на этом потоке встанут.

// в async-обработчике — блокирует worker
std::thread::sleep(std::time::Duration::from_secs(1));

// корректная пауза в async
tokio::time::sleep(std::time::Duration::from_secs(1)).await;

Разбор:

std::thread::sleep блокирует поток ОС целиком, поэтому worker Tokio временно перестаёт обслуживать другие задачи.
tokio::time::sleep(...).await ставит async-задачу на паузу неблокирующим способом.
Во время await runtime переключается на другие futures, что повышает пропускную способность сервиса.
Для async-кода правило простое: блокирующие операции из std::thread заменять на async-эквиваленты Tokio.
Такой выбор напрямую влияет на latency под нагрузкой.

Синхронная библиотека (сжатие, старый SQLite, блокирующий HTTP):

let result = tokio::task::spawn_blocking(|| heavy_sync_work())
    .await
    .expect("join spawn_blocking")?;

Разбор:

spawn_blocking запускает тяжёлую синхронную функцию в отдельном пуле потоков.
Это изолирует блокирующую работу от основных async-worker'ов.
.await ждёт завершения задачи и возвращает её результат обратно в async-контекст.
expect("join spawn_blocking") обрабатывает возможный JoinError (например, panic внутри задачи).
Оператор ? затем пробрасывает уже бизнес-ошибку heavy_sync_work() вызывающему коду.

spawn_blocking берёт поток из отдельного пула; основной runtime продолжает обслуживать I/O. Тяжёлую работу выносите туда осознанно, пул ограничен.

Типичные ошибки

Дедлок — поток A ждёт lock X, поток B ждёт lock Y, каждый держит другой lock. Решение — один порядок захвата, меньше вложенных Mutex, чаще каналы.
Долгий I/O под Mutex — остальные потоки простаивают. Копируйте нужные данные, отпускайте guard, затем сетевой вызов.
thread::sleep в async — "замораживает" воркер Tokio.
Паника в spawn без join — ошибка теряется; в серверах логируют результат join или используют обёртки.

Выбор примитивов под задачу

Удобно держать короткое правило:

передаёте независимые сообщения — берите канал;
делите изменяемое состояние — Arc<Mutex<T>> или Arc<RwLock<T>>;
нужен только счётчик — рассмотрите атомики (AtomicUsize, AtomicBool);
CPU-задачи "распараллелить по массиву" — часто быстрее и проще через rayon.

Для backend-кода комбинация "канал + worker pool" обычно масштабируется лучше, чем один глобальный Mutex на всё приложение.

Паттерн producer-consumer

Классическая схема:

несколько producer-потоков читают данные (файлы, очередь задач, сеть);
отправляют сообщения в mpsc;
один consumer-поток сериализует запись в БД или файл.

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

Graceful shutdown для потоков

Потоки завершать удобно "мягко", через служебное сообщение:

enum Cmd {
    Work(u64),
    Stop,
}

Разбор:

enum Cmd задаёт явный протокол сообщений для worker-потока.
Вариант Work(u64) несёт полезную нагрузку задачи (например, id или размер работы).
Вариант Stop служит сигналом корректного завершения цикла обработки.
Такой подход делает управление жизненным циклом потока декларативным и читаемым.
С enum-командами проще расширять протокол новыми типами действий без магических строк.

Worker читает канал в цикле и при Stop выходит. Такой подход лучше, чем "бесконечный цикл + принудительное завершение процесса", потому что позволяет корректно закрыть файлы и доотправить логи.

Полный цикл worker с командами:

fn worker(rx: mpsc::Receiver<Cmd>) {
    while let Ok(cmd) = rx.recv() {
        match cmd {
            Cmd::Work(id) => println!("обработка {id}"),
            Cmd::Stop => break,
        }
    }
    println!("worker завершён");
}

Разбор:

while let Ok(cmd) = rx.recv() читает команды, пока канал открыт.
Cmd::Work(id) обрабатывает полезную задачу.
Cmd::Stop завершает цикл без принудительного kill потока.
После break можно освободить ресурсы и записать финальный лог.
Такой shutdown предсказуемее для сервисов и CLI-утилит.

Производительность и диагностика

держите критическую секцию под Mutex минимальной;
измеряйте время на lock() в профилировщике, а не "на глаз";
при высокой конкуренции делите состояние на несколько lock-объектов;
добавляйте метрики длины очередей каналов и времени обработки задач.

Эти сигналы быстро показывают, где узкое место: CPU, ожидание lock или медленный I/O.

Связанные материалы

Асинхронное программирование
Важные трейты и типы — Arc, Mutex, Rc
Системное программирование
Тестирование
Cargo workspace

Основа по протоколу

Базовый разбор HTTP и HTTPS находится в отдельной статье — HTTP как основа веб-интеграций.

Потоки и синхронные каналы в Rust