Tokio: как отправить и принять множество событий

Вступление

#

В прошлой главе (Каналы в Rust при поддержке Tokio) мы разобрались, как работают каналы, и помогли роботу RUST-Y перевести связь корабля Vectoria на асинхронный tokio::sync::mpsc, избавившись от опасных блокировок главного потока.

Теперь перед экипажем стоит критически важная задача: состыковаться с дрейфующим медицинским модулем и спасти находящегося там доктора Арчи. Чтобы сделать это безопасно, бортовым системам нужно обрабатывать множество событий одновременно — принимать телеметрию датчиков, слушать ручные команды капитана Новы с мостика и следить за таймером аварийного отключения.

В этой главе мы поможем RUST-Y построить настоящую асинхронную систему обработки событий на базе Tokio. Мы разберем:

• Как одновременно слушать разные асинхронные источники с помощью макроса tokio::select!.
• Как организовать рассылку уведомлений всем системам сразу с помощью tokio::sync::broadcast.
• Как управлять пулом динамически запускаемых задач и обрабатывать их результаты через tokio::task::JoinSet.

Пролог. Сближение

#

Медицинский модуль медленно вращался в холодном свете планеты Лира-Бета. Защитный контур модуля («Страж») перешел в аварийный режим и блокировал любые попытки внешней стыковки, считая их угрозой.

На мостике Vectoria Капитан Нова внимательно следил за показателями сближения.

— «Зори, какая дистанция?» — спросил Нова. — «Триста метров. Сближение стабильное, но если автоматика модуля посчитает наш маневр агрессивным, она задействует магнитные выталкиватели. Мы потеряем контакт навсегда.» — «RUST-Y!» — капитан повернулся к роботу. — «Нам нужен надежный диспетчер событий. Мы должны непрерывно считывать телеметрию сближения, слушать ручные команды отмены от Зори и контролировать таймер безопасности. Если телеметрия покажет перегрузку или придет команда отмены — мы должны среагировать мгновенно.»

RUST-Y бодро защелкал реле: — «Капитан! Я переписал код. Больше никаких блокировок потока! Я использую асинхронные каналы Tokio. Но как мне объединить их все в один цикл обработки?»

Нова улыбнулся и указал на голографический экран: — «Для этого у нас есть макрос tokio::select!. Он работает как стрелочник на железной дороге — ждет, какая ветка сработает первой, и моментально передает управление туда. Давай посмотрим.»

Часть 1. Мультиплексирование и макрос tokio::select!

#

Когда нам нужно ожидать выполнения нескольких асинхронных операций одновременно, мы используем tokio::select!. Этот макрос принимает список выражений (Futures) и выполняет ветку той Future, которая завершилась первой.

Давайте посмотрим на схему, которую спроектировал RUST-Y для управления стыковкой:

 1// ?hidden:start
 2use tokio::sync::mpsc;
 3use tokio::time::{sleep, Duration, interval};
 4
 5#[derive(Debug)]
 6enum Command {
 7    AdjustShields(u32),
 8    Evacuate,
 9}
10
11#[derive(Debug)]
12enum Telemetry {
13    Temperature(f32),
14    ReactorStatus(bool),
15}
16// ?hidden:end
17
18#[tokio::main]
19async fn main() {
20    let (cmd_tx, mut cmd_rx) = mpsc::channel(10);
21    let (telemetry_tx, mut telemetry_rx) = mpsc::channel(10);
22
23    // Имитируем отправку команд в фоне
24    tokio::spawn(async move {
25        sleep(Duration::from_millis(150)).await;
26        let _ = cmd_tx.send(Command::AdjustShields(80)).await;
27        sleep(Duration::from_millis(250)).await;
28        let _ = cmd_tx.send(Command::Evacuate).await;
29    });
30
31    // Имитируем отправку телеметрии в фоне
32    tokio::spawn(async move {
33        let _ = telemetry_tx.send(Telemetry::Temperature(36.6)).await;
34        sleep(Duration::from_millis(200)).await;
35        let _ = telemetry_tx.send(Telemetry::ReactorStatus(true)).await;
36    });
37
38    let mut heartbeat = interval(Duration::from_millis(100));
39    // Пропустим первый мгновенный тик, чтобы не засорять вывод
40    heartbeat.tick().await;
41
42    loop {
43        tokio::select! {
44            // Читаем команды
45            Some(cmd) = cmd_rx.recv() => {
46                println!("[МОСТИК] Получена команда: {:?}", cmd);
47                if let Command::Evacuate = cmd {
48                    println!("[МОСТИК] Внимание! Начинаем эвакуацию!");
49                    break;
50                }
51            }
52            // Читаем телеметрию
53            Some(tel) = telemetry_rx.recv() => {
54                println!("[МОСТИК] Телеметрия: {:?}", tel);
55            }
56            // Периодический сигнал
57            _ = heartbeat.tick() => {
58                println!("[МОСТИК] Системы жизнеобеспечения в норме...");
59            }
60        }
61    }
62}
63

Как это устроено под капотом?

#

tokio::select! {
    Some(cmd) = cmd_rx.recv() => { ... }
    Some(tel) = telemetry_rx.recv() => { ... }
    _ = heartbeat.tick() => { ... }
}

1. Ожидание первого события: tokio::select! опрашивает все указанные ветки. Как только в канале cmd_rx появляется команда, или в telemetry_rx приходит значение, или срабатывает таймер heartbeat.tick(), макрос выполняет соответствующий блок кода.
2. Безопасность при отмене (Cancellation Safety): Это ключевое понятие асинхронного Rust. Когда одна из ветвей tokio::select! завершается, все остальные ветви уничтожаются (dropped). Это означает, что невыбранные Future прекращают свое существование.
   • mpsc::Receiver::recv является безопасным к отмене: если мы удалим Future ожидания сообщения, само сообщение в канале не потеряется — оно останется в буфере до следующей попытки чтения.
   • Внимание: Не все операции безопасны к отмене! Например, чтение из сетевого сокета tokio::io::AsyncReadExt::read при отмене может привести к потере части уже прочитанных байт.
3. Защита от голодания (Starvation): Если бы select! всегда проверял ветки по порядку (сверху вниз), то при высокой частоте сообщений в первом канале до второго канала очередь могла бы никогда не дойти. Чтобы этого избежать, Tokio при одновременной готовности нескольких веток выбирает одну из них псевдослучайным образом.

Часть 2. Оповещение всех систем через tokio::sync::broadcast

#

Стыковка подходила к финальной фазе. Внезапно датчики зафиксировали вспышку на поверхности планеты — в сторону модуля и Vectoria двигалось облако ионизированной плазмы.

— «Капитан! Нам нужно объявить тревогу и немедленно передать команду подготовки к удару на все модули корабля!» — крикнул Зори. — «RUST-Y, отправь сигнал экстренного старта!» — скомандовал Нова.

Робот засуетился: — «Но капитан! Наш канал mpsc работает по принципу «один отправитель — один получатель». Если я отправлю команду в mpsc, ее получит либо модуль жизнеобеспечения, либо двигатели, но не оба сразу! Нам нужно разослать сообщение абсолютно всем!»

— «Для этого используй broadcast канал», — спокойно ответил Нова. — «Он работает как радиостанция: один передает, а слушают все, кто настроен на эту волну.»

 1// ?hidden:start
 2use tokio::sync::broadcast;
 3use tokio::time::{sleep, Duration};
 4// ?hidden:end
 5
 6#[tokio::main]
 7async fn main() {
 8    // Создаем broadcast-канал с буфером на 2 сообщения
 9    let (tx, mut rx1) = broadcast::channel::<String>(2);
10    let mut rx2 = tx.subscribe();
11
12    // Первый получатель: Модуль жизнеобеспечения (быстрый)
13    tokio::spawn(async move {
14        loop {
15            match rx1.recv().await {
16                Ok(msg) => println!("[ЖИЗНЕОБЕСПЕЧЕНИЕ] Получено: {}", msg),
17                Err(broadcast::error::RecvError::Closed) => break,
18                Err(e) => println!("[ЖИЗНЕОБЕСПЕЧЕНИЕ] Ошибка: {:?}", e),
19            }
20        }
21    });
22
23    // Второй получатель: Двигательный отсек (медленный)
24    tokio::spawn(async move {
25        loop {
26            // Имитируем долгую обработку, из-за чего буфер переполнится
27            sleep(Duration::from_millis(300)).await;
28            match rx2.recv().await {
29                Ok(msg) => println!("[ДВИГАТЕЛИ] Получено: {}", msg),
30                Err(broadcast::error::RecvError::Lagged(n)) => {
31                    println!("[ДВИГАТЕЛИ] Отстали! Пропущено {} сообщений.", n);
32                }
33                Err(broadcast::error::RecvError::Closed) => break,
34            }
35        }
36    });
37
38    // Отправляем сообщения с мостика
39    sleep(Duration::from_millis(50)).await;
40    let _ = tx.send("Сигнал 1: Прогрев систем".to_string());
41    let _ = tx.send("Сигнал 2: Калибровка щитов".to_string());
42    let _ = tx.send("Сигнал 3: Запуск реактора".to_string());
43    let _ = tx.send("Сигнал 4: Экстренный старт!".to_string());
44
45    // Даем время задачам завершиться
46    sleep(Duration::from_millis(1000)).await;
47}
48

Особенности канала broadcast:

#

// Каждая подписка создает новый приемник:
let mut rx = tx.subscribe();

// При чтении медленный получатель отловит ошибку отставания:
Err(RecvError::Lagged(n)) => {
    println!("Пропущено {} сообщений!", n);
}

• Схема Multi-producer Multi-consumer (MPMC): Любое сообщение, отправленное через Sender, клонируется и доставляется всем активным Receiver. Получатели создаются путем вызова метода tx.subscribe() на строке 4 .
• Проблема медленного получателя (Lagging): Поскольку буфер канала имеет фиксированный размер (в нашем примере — 2 сообщения), быстрый отправитель может переполнить его, если один из получателей работает медленно.
• В нашем коде Двигательный отсек имитирует долгую работу через sleep. Когда буфер переполняется, медленный получатель при очередной попытке чтения получит ошибку RecvError::Lagged(n) на строке 5 . Это сигнализирует о том, что n сообщений были пропущены, так как кольцевой буфер перезаписал старые данные. Получатель должен корректно обработать эту ошибку и продолжить чтение актуальных данных.

Часть 3. Динамические задачи и tokio::task::JoinSet

#

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

Количество зондов может меняться в зависимости от плотности космического мусора.

— «Если мы запустим зонды и будем ждать их через join_all, нам придется ждать завершения работы последнего, даже если первый найдет челнок за секунду», — заметил Зори. — «Верно», — согласился Нова. — «Нам нужен инструмент, который позволит собирать результаты по мере их готовности. И как только цель будет найдена — мы должны немедленно отозвать (отменить) все остальные зонды, чтобы сберечь энергию.»

RUST-Y открыл терминал: — «Я знаю! Мы используем tokio::task::JoinSet!»

 1// ?hidden:start
 2use tokio::task::JoinSet;
 3use tokio::time::{sleep, Duration};
 4// ?hidden:end
 5
 6async fn scan_sector(sector_id: u32) -> (u32, &'static str) {
 7    // Разное время сканирования секторов для имитации асинхронности
 8    let delay = match sector_id {
 9        1 => 150,
10        2 => 50,
11        3 => 250,
12        _ => 100,
13    };
14    sleep(Duration::from_millis(delay)).await;
15    
16    match sector_id {
17        2 => (sector_id, "Обнаружен спасательный челнок доктора!"),
18        _ => (sector_id, "Сектор пуст"),
19    }
20}
21
22#[tokio::main]
23async fn main() {
24    let mut set = JoinSet::new();
25
26    // Запускаем 3 задачи сканирования секторов динамически
27    for sector_id in 1..=3 {
28        set.spawn(scan_sector(sector_id));
29    }
30
31    // Обрабатываем результаты по мере их завершения
32    while let Some(res) = set.join_next().await {
33        match res {
34            Ok((sector, status)) => {
35                println!("[ЗОНД] Сектор {} завершил сканирование: {}", sector, status);
36                if status.contains("челнок") {
37                    println!("[МОСТИК] Цель найдена! Прекращаем сканирование остальных секторов.");
38                    set.abort_all(); // Отменяем все оставшиеся задачи
39                    break;
40                }
41            }
42            Err(e) => println!("[ОШИБКА] Задача прервана: {:?}", e),
43        }
44    }
45}
46

Почему JoinSet — это отличное решение?

#

// Запускаем асинхронную задачу в пул JoinSet:
set.spawn(async move { ... });

// Получаем первый готовый результат:
if let Some(res) = set.join_next().await {
    // Отменяем все остальные задачи:
    set.abort_all();
}

1. Динамический пул задач: Мы можем добавлять новые задачи (Future) прямо во время работы пула с помощью set.spawn().
2. Получение результатов по готовности: Метод set.join_next().await возвращает результат первой завершившейся задачи. Нам не нужно ждать все остальные.
3. Мгновенная отмена через abort_all(): Если мы нашли то, что искали, вызов set.abort_all() отправляет сигнал отмены (cancellation) всем оставшимся задачам. Это изящно решает проблему утечки ресурсов и лишней работы.

Эпилог. Доктор на борту

#

Зонд №2 прислал координаты спасательного челнока. RUST-Y моментально отменил остальные задачи, а гравитационный луч Vectoria бережно притянул челнок к основному шлюзу.

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

— «Значит так», — подытожил RUST-Y. — «Если нужно объединить несколько разных источников в один поток — используем tokio::select!. Если нужно разослать сообщение всем системам корабля — берем broadcast. А если запускаем кучу однотипных зондов и хотим гибко ими управлять — нет ничего лучше JoinSet.»

Нова одобрительно кивнул: — «Именно так. Главное — понимать сильные стороны каждого инструмента и не забывать про безопасность отмены.»

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

#

Давайте проверим, насколько хорошо вы разобрались с отправкой и приемом множества событий в Tokio.