Каналы в Rust при поддержке Tokio

Вступление

#

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

Хочется, чтобы этот обмен был простым, безопасным и предсказуемым. Для этого в Rust существуют каналы. Они позволяют отделить тех, кто отправляет данные (producers), от того, кто принимает и обрабатывает их (consumer).

В этой главе мы снова поднимемся на борт корабля Vectoria. Капитан Нова и робот RUST-Y окажутся в ситуации, где от правильного выбора канала зависит успех спасательной операции.

Вместе с героями мы разберём:

• чем std::sync::mpsc отличается от tokio::sync::mpsc,
• почему блокирующий recv() может остановить выполнение всей программы,
• как на практике использовать модель «много отправителей — один получатель».

Пролог. Сигнал

#

После прохождения через пространственные врата корабль Vectoria дрейфовал у планеты Лира-Бета.

На низкой орбите находился поврежденный медицинский модуль, подающий слабый аварийный сигнал.

 1// ?hidden:start
 2use std::sync::mpsc;
 3use std::thread;
 4use std::time::Duration;
 5// ?hidden:end
 6
 7fn main() {
 8    let (tx, rx) = mpsc::channel();
 9
10    // Запускаем поток, имитирующий долгую работу навигатора
11    thread::spawn(move || {
12        println!("[НАВИГАЦИЯ] Выполняем расчет траектории...");
13        thread::sleep(Duration::from_millis(500));
14        let _ = tx.send("Траектория рассчитана!".to_string());
15    });
16
17    println!("[МОСТИК] Ждем данные от навигатора...");
18    
19    // Внимание: recv() БЛОКИРУЕТ текущий поток!
20    // Если этот поток управляет всем интерфейсом или рантаймом,
21    // вся программа зависнет в ожидании.
22    match rx.recv() {
23        Ok(msg) => println!("[МОСТИК] Успех: {}", msg),
24        Err(e) => println!("[МОСТИК] Ошибка связи: {:?}", e),
25    }
26
27    println!("[МОСТИК] Работа продолжается.");
28}
29

Капитан Нова выделил строку с вызовом recv():

// Поток зависает здесь, если сообщений нет:
let msg = rx.recv().unwrap();

— «Вот здесь всё и останавливается. Метод recv() в синхронном канале std::sync::mpsc блокирует текущий поток операционной системы до тех пор, пока не придет сообщение. В простом многопоточном приложении это нормально — один поток засыпает в ожидании, пока другой работает.»

Он переключил схему на общий контур управления кораблем: — «Но у нас однопоточный асинхронный контекст. Когда ты вызываешь блокирующий recv() в основном потоке воркера, ты говоришь планировщику: “Ничего не делай, пока не ответит навигатор”. И пока он рассчитывает траекторию, весь корабль замирает: мы не можем ни щиты скорректировать, ни принять экстренную команду отмены.»

Доктор Арчи добавил: — «Если внутри такой блокировки зависнет система мониторинга состояния жизнеобеспечения, мы узнаем об аварии слишком поздно. Блокирующий вызов в асинхронном коде превращается в угрозу.»

Часть 2. Асинхронные каналы Tokio

#

— «Решение не в том, чтобы отказаться от каналов», — продолжит Нова. — «Нам просто нужен инструмент, созданный специально для асинхронной среды. Такой, который не блокирует поток ОС, а приостанавливает только текущую задачу, уступая время другим.»

Он переписал код RUST-Y с использованием tokio::sync::mpsc:

 1// ?hidden:start
 2use tokio::sync::mpsc;
 3use tokio::time::{sleep, Duration};
 4// ?hidden:end
 5
 6#[tokio::main]
 7async fn main() {
 8    // Создаем асинхронный mpsc-канал с емкостью буфера на 10 элементов
 9    let (tx, mut rx) = mpsc::channel(10);
10
11    // Запускаем асинхронную задачу в Tokio
12    tokio::spawn(async move {
13        println!("[НАВИГАЦИЯ] Начинаем расчет...");
14        sleep(Duration::from_millis(300)).await;
15        let _ = tx.send("Траектория рассчитана асинхронно!".to_string()).await;
16    });
17
18    println!("[МОСТИК] Ожидаем данные (неблокирующий await)...");
19
20    // recv() возвращает Future. Она приостанавливает выполнение нашей задачи,
21    // но воркер-поток Tokio продолжает выполнять другие задачи в это время!
22    if let Some(msg) = rx.recv().await {
23        println!("[МОСТИК] Принято: {}", msg);
24    }
25
26    println!("[МОСТИК] Сближение завершено.");
27}
28

Почему это сработало?

#

// Асинхронное чтение освобождает поток планировщика:
let msg = rx.recv().await; 
    2

1. Неблокирующий recv().await: В асинхронном канале вызов recv() возвращает Future. Когда мы пишем .await на строке 2 , планировщик Tokio видит, что данных в канале еще нет. Он временно «замораживает» текущую задачу и переключает поток ОС на выполнение других задач. Как только данные приходят, планировщик возобновляет выполнение.
2. Ограниченная емкость (Bounded Channel): Асинхронный канал Tokio создается через mpsc::channel(capacity). Емкость буфера обязательна! Если отправитель шлет сообщения быстрее, чем получатель успевает их читать, буфер переполнится, и вызов send().await приостановит задачу отправителя. Это называется backpressure (обратное давление) и защищает приложение от утечки памяти.

Часть 3. Модель «Многие отправители — один получатель»

#

— «Капитан, а если мне нужно собирать телеметрию сразу с нескольких сенсорных модулей?» — спросил RUST-Y. — «Мне создавать канал для каждого?»

— «Нет», — ответил Зори. — «Канал называется mpsc — Multi-producer Single-consumer. Это значит, что отправителей может быть много, а получатель — всегда один. Мы можем просто клонировать передатчик и раздать его разным задачам.»

RUST-Y обновил код, применив клонирование отправителя:

 1// ?hidden:start
 2use tokio::sync::mpsc;
 3use tokio::time::{sleep, Duration};
 4// ?hidden:end
 5
 6async fn sensor_task(id: u32, tx: mpsc::Sender<String>) {
 7    for i in 1..=3 {
 8        sleep(Duration::from_millis(50 * id as u64)).await;
 9        let msg = format!("Датчик {} -> Значение {}", id, i);
10        let _ = tx.send(msg).await;
11    }
12}
13
14#[tokio::main]
15async fn main() {
16    let (tx, mut rx) = mpsc::channel(10);
17
18    // Запускаем два параллельных датчика, каждый со своей копией Sender
19    tokio::spawn(sensor_task(1, tx.clone()));
20    tokio::spawn(sensor_task(2, tx.clone()));
21
22    // Важно: удаляем исходный Sender, иначе цикл rx.recv() никогда не завершится,
23    // так как канал будет считаться открытым
24    drop(tx);
25
26    println!("[ЦЕНТР] Ожидаем показания датчиков...");
27
28    // Канал закроется автоматически, когда все отправители (Sender) будут удалены
29    while let Some(msg) = rx.recv().await {
30        println!("[ЦЕНТР] Получено: {}", msg);
31    }
32
33    println!("[ЦЕНТР] Все датчики отключены.");
34}
35

Важные детали работы с клонами Sender:

#

// Освобождаем исходный передатчик:
drop(tx); 
    3


// Получатель завершит цикл, когда все Sender исчезнут:
while let Some(msg) = rx.recv().await { 
    4

    println!("Получено: {:?}", msg);
}

• Метод tx.clone(): Мы можем создавать сколько угодно копий отправителя и передавать их в асинхронные задачи.
• Закрытие канала: Канал закрывается автоматически, когда все экземпляры Sender (включая самый первый, оригинальный) удаляются из памяти (drop).
• Явный drop(tx) в main: Обратите внимание на строку drop(tx) на строке 3 в основной задаче после запуска рабочих потоков. Если мы забудем удалить оригинальный tx в main, то получатель Receiver в цикле while let Some(...) на строке 4 зависнет навсегда. Канал будет считаться открытым, так как одна копия Sender все еще находится на мостике, пусть из нее никто ничего и не отправляет.

Эпилог

#

— «Кажется, теперь все сходится!» — обрадовался RUST-Y. — «Мы настроили асинхронный канал с датчиков сближения на мостик, поток не блокируется, а исходный отправитель аккуратно удаляется после запуска.»

— «Отлично», — подытожил капитан Нова. — «Связь установлена. Зори, начинаем маневр сближения с модулем доктора Арчи.»

Корабль Vectoria мягко скользнул вперед, а его асинхронные системы работали слаженно и без блокировок.

Проверь свои знания!

#

Пройдите небольшой тест, чтобы закрепить понимание работы асинхронных каналов в Rust и Tokio.