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

> Продолжение приключений экипажа Vectoria — разбор способов отправки и обработки множества событий в Tokio: использование макроса tokio::select!, широковещательных каналов broadcast и динамического пула задач с помощью JoinSet.

- **URL:** https://eagle2000.space/rust/tokio-kak-otpravit-i-prinyat-mnozhestvo-sobytij/
- **Date:** 2026-08-14

- **Tags:** rust, dev

---## Вступление

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

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

В этой главе мы поможем RUST-Y построить настоящую асинхронную систему обработки событий на базе **Tokio**. Мы разберем:
- Как одновременно слушать разные асинхронные источники с помощью макроса `tokio::select!`.
- Как организовать рассылку уведомлений всем системам сразу с помощью `tokio::sync::broadcast`.
- Как управлять пулом динамически запускаемых задач и обрабатывать их результаты через `tokio::task::JoinSet`.

{{< figure src="diagram.png" alt="Схемы работы select!, broadcast и JoinSet" >}}

{{< alert "comment" >}}
В конце статьи вас ждет интерактивный мини-квиз. Внимательно разбирайте примеры кода, чтобы без ошибок ответить на вопросы по `select!`, `broadcast` и `JoinSet`.
{{< /alert >}}

---

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

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

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

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

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

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

---

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

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

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


### 💻 Слушаем несколько каналов с tokio::select!


```rust
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration, interval};

#[derive(Debug)]
enum Command {
    AdjustShields(u32),
    Evacuate,
}

#[derive(Debug)]
enum Telemetry {
    Temperature(f32),
    ReactorStatus(bool),
}
#[tokio::main]
async fn main() {
    let (cmd_tx, mut cmd_rx) = mpsc::channel(10);
    let (telemetry_tx, mut telemetry_rx) = mpsc::channel(10);

    // Имитируем отправку команд в фоне
    tokio::spawn(async move {
        sleep(Duration::from_millis(150)).await;
        let _ = cmd_tx.send(Command::AdjustShields(80)).await;
        sleep(Duration::from_millis(250)).await;
        let _ = cmd_tx.send(Command::Evacuate).await;
    });

    // Имитируем отправку телеметрии в фоне
    tokio::spawn(async move {
        let _ = telemetry_tx.send(Telemetry::Temperature(36.6)).await;
        sleep(Duration::from_millis(200)).await;
        let _ = telemetry_tx.send(Telemetry::ReactorStatus(true)).await;
    });

    let mut heartbeat = interval(Duration::from_millis(100));
    // Пропустим первый мгновенный тик, чтобы не засорять вывод
    heartbeat.tick().await;

    loop {
        tokio::select! {
            // Читаем команды
            Some(cmd) = cmd_rx.recv() => {
                println!("[МОСТИК] Получена команда: {:?}", cmd);
                if let Command::Evacuate = cmd {
                    println!("[МОСТИК] Внимание! Начинаем эвакуацию!");
                    break;
                }
            }
            // Читаем телеметрию
            Some(tel) = telemetry_rx.recv() => {
                println!("[МОСТИК] Телеметрия: {:?}", tel);
            }
            // Периодический сигнал
            _ = heartbeat.tick() => {
                println!("[МОСТИК] Системы жизнеобеспечения в норме...");
            }
        }
    }
}
```

**Мультиплексирование через tokio::select!**
- Слушаем несколько источников событий одновременно.
- Ветки выбираются по мере готовности данных.
- Работает для каналов, таймеров и любых Future.


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

```rust
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` канал», — спокойно ответил Нова. — «Он работает как радиостанция: один передает, а слушают все, кто настроен на эту волну.»


### 💻 Оповещение систем через broadcast-канал


```rust
use tokio::sync::broadcast;
use tokio::time::{sleep, Duration};
#[tokio::main]
async fn main() {
    // Создаем broadcast-канал с буфером на 2 сообщения
    let (tx, mut rx1) = broadcast::channel::<String>(2);
    let mut rx2 = tx.subscribe();

    // Первый получатель: Модуль жизнеобеспечения (быстрый)
    tokio::spawn(async move {
        loop {
            match rx1.recv().await {
                Ok(msg) => println!("[ЖИЗНЕОБЕСПЕЧЕНИЕ] Получено: {}", msg),
                Err(broadcast::error::RecvError::Closed) => break,
                Err(e) => println!("[ЖИЗНЕОБЕСПЕЧЕНИЕ] Ошибка: {:?}", e),
            }
        }
    });

    // Второй получатель: Двигательный отсек (медленный)
    tokio::spawn(async move {
        loop {
            // Имитируем долгую обработку, из-за чего буфер переполнится
            sleep(Duration::from_millis(300)).await;
            match rx2.recv().await {
                Ok(msg) => println!("[ДВИГАТЕЛИ] Получено: {}", msg),
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    println!("[ДВИГАТЕЛИ] Отстали! Пропущено {} сообщений.", n);
                }
                Err(broadcast::error::RecvError::Closed) => break,
            }
        }
    });

    // Отправляем сообщения с мостика
    sleep(Duration::from_millis(50)).await;
    let _ = tx.send("Сигнал 1: Прогрев систем".to_string());
    let _ = tx.send("Сигнал 2: Калибровка щитов".to_string());
    let _ = tx.send("Сигнал 3: Запуск реактора".to_string());
    let _ = tx.send("Сигнал 4: Экстренный старт!".to_string());

    // Даем время задачам завершиться
    sleep(Duration::from_millis(1000)).await;
}
```

**Канал tokio::sync::broadcast**
- Отправка сообщений по схеме "многие ко многим" (Multi-producer Multi-consumer).
- Если буфер переполняется из-за медленного получателя, Tokio выбрасывает `RecvError::Lagged`.
- Идеально подходит для общих событий, таких как сигналы остановки или глобальные уведомления.


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

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

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

- **Схема Multi-producer Multi-consumer (MPMC):** Любое сообщение, отправленное через `Sender`, клонируется и доставляется всем активным `Receiver`. Получатели создаются путем вызова метода `tx.subscribe()` на строке {{< codelink id="a4" />}}.
- **Проблема медленного получателя (Lagging):** Поскольку буфер канала имеет фиксированный размер (в нашем примере — 2 сообщения), быстрый отправитель может переполнить его, если один из получателей работает медленно. 
- В нашем коде Двигательный отсек имитирует долгую работу через `sleep`. Когда буфер переполняется, медленный получатель при очередной попытке чтения получит ошибку `RecvError::Lagged(n)` на строке {{< codelink id="a5" />}}. Это сигнализирует о том, что `n` сообщений были пропущены, так как кольцевой буфер перезаписал старые данные. Получатель должен корректно обработать эту ошибку и продолжить чтение актуальных данных.

---

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

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

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

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

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


### 💻 Поиск челнока через JoinSet


```rust
use tokio::task::JoinSet;
use tokio::time::{sleep, Duration};
async fn scan_sector(sector_id: u32) -> (u32, &'static str) {
    // Разное время сканирования секторов для имитации асинхронности
    let delay = match sector_id {
        s if s == 1 => 150,
        s if s == 2 => 50,
        s if s == 3 => 250,
        _ => 100,
    };
    sleep(Duration::from_millis(delay)).await;
    
    match sector_id {
        s if s == 2 => (sector_id, "Обнаружен спасательный челнок доктора!"),
        _ => (sector_id, "Сектор пуст"),
    }
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();

    // Запускаем 3 задачи сканирования секторов динамически
    for sector_id in 1..=3 {
        set.spawn(scan_sector(sector_id));
    }

    // Обрабатываем результаты по мере их завершения
    while let Some(res) = set.join_next().await {
        match res {
            Ok((sector, status)) => {
                println!("[ЗОНД] Сектор {} завершил сканирование: {}", sector, status);
                if status.contains("челнок") {
                    println!("[МОСТИК] Цель найдена! Прекращаем сканирование остальных секторов.");
                    set.abort_all(); // Отменяем все оставшиеся задачи
                    break;
                }
            }
            Err(e) => println!("[ОШИБКА] Задача прервана: {:?}", e),
        }
    }
}
```

**Менеджер задач tokio::task::JoinSet**
- Динамический пул конкурентных задач, возвращающий результаты по мере их готовности.
- Позволяет легко прервать все оставшиеся задачи через `set.abort_all()`.
- Решает проблему утечки ресурсов при отмене и упрощает сбор результатов по сравнению с `join_all`.


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

```rust
// Запускаем асинхронную задачу в пул 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.


### 📝 Tokio: Отправка и прием событий


**1. Что происходит с невыбранными ветками (Futures) в tokio::select! после того, как одна из веток завершилась?**

- [ ] Они продолжают выполняться в фоне как фоновые задачи
- [ ] Они отменяются путем удаления (drop) соответствующей Future
- [ ] Они приостанавливаются и ждут следующего вызова select!
- [ ] Происходит паника рантайма

**2. Как ведет себя tokio::select!, если несколько веток готовы к выполнению одновременно?**

- [ ] Всегда выбирает первую ветку по порядку в коде
- [ ] Выбирает одну случайную ветку для предотвращения голодания (starvation)
- [ ] Запускает все готовые ветки параллельно
- [ ] Возвращает ошибку компиляции

**3. Какое поведение демонстрирует канал tokio::sync::broadcast при отставании одного из получателей?**

- [ ] Медленный получатель блокирует отправку сообщений для всех остальных
- [ ] Канал автоматически увеличивает размер буфера до бесконечности
- [ ] Медленный получатель получает ошибку RecvError::Lagged(n) и пропускает n сообщений
- [ ] Программа завершается с паникой

**4. В какой ситуации предпочтительнее использовать broadcast-канал вместо mpsc?**

- [ ] Когда нужно передать сообщение ровно одному свободному воркеру
- [ ] Когда одно и то же сообщение (например, команду сброса или смены статуса) должны получить все подписчики
- [ ] Когда нужно передавать данные между синхронным и асинхронным кодом
- [ ] Когда требуется максимальная пропускная способность для одного получателя

**5. В чем главное преимущество tokio::task::JoinSet по сравнению с futures::future::join_all при работе с множеством динамических задач?**

- [ ] JoinSet работает быстрее, так как не использует выделение памяти
- [ ] JoinSet позволяет обрабатывать результаты задач по мере их готовности и легко отменять оставшиеся через abort_all()
- [ ] JoinSet поддерживает работу только в однопоточном режиме
- [ ] JoinSet не требует импорта сторонних крейтов